diff --git a/.github/workflows/packages.yml b/.github/workflows/packages.yml index b83f60344..5670026a0 100644 --- a/.github/workflows/packages.yml +++ b/.github/workflows/packages.yml @@ -549,6 +549,75 @@ jobs: sys.exit(1 if bad else 0) PY + # Real native list apps read through the row census (identical rows halt, + # record ids pick the row, a unique row runs): a WPF ListBox over UI + # Automation on Windows and a Swift NSTableView over AX on macOS. Both legs + # must run every real-app test; the step summary lists each outcome. + flow-native-census: + name: flow native row census (${{ matrix.os }}) + needs: scope + if: needs.scope.outputs.flow == 'true' + runs-on: ${{ matrix.os }} + timeout-minutes: 25 + strategy: + fail-fast: false + matrix: + os: [windows-latest, macos-latest] + defaults: + run: + working-directory: packages/flow + shell: bash + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + cache: pip + - name: Install + run: | + pip install -e ../types -e . pytest pytest-timeout + if [ "$RUNNER_OS" = "Windows" ]; then pip install uiautomation; fi + if [ "$RUNNER_OS" = "macOS" ]; then + pip install pyobjc-framework-ApplicationServices pyobjc-framework-Cocoa pyobjc-framework-Quartz + fi + - name: Census unit tests and real-app tests + env: + OPENADAPT_TEST_WINDOWS_UIA: "1" + OPENADAPT_TEST_MACOS_AX: "1" + run: | + mkdir -p runs + pytest -v -rs tests/test_native_identity_census.py \ + tests/test_native_census_real_apps.py --junitxml=runs/native-junit.xml + - name: The real-app tests ran (no skips) + run: | + python - <<'PY' + import os + import sys + import xml.etree.ElementTree as ET + prefix = {"Windows": "test_windows_wpf_", "macOS": "test_macos_table_"}[ + os.environ["RUNNER_OS"] + ] + cases = [ + c for c in ET.parse("runs/native-junit.xml").getroot().iter("testcase") + if c.get("name", "").startswith(prefix) + ] + lines = [f"### Native row census, real app ({os.environ['RUNNER_OS']})", ""] + for case in cases: + skip = case.find("skipped") + failed = case.find("failure") is not None or case.find("error") is not None + state = "skipped: " + (skip.get("message") or "") if skip is not None else ( + "failed" if failed else "passed" + ) + lines.append(f"- `{case.get('name')}`: {state}") + with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as out: + out.write("\n".join(lines) + "\n") + print("\n".join(lines)) + skipped = [c.get("name") for c in cases if c.find("skipped") is not None] + sys.exit(1 if not cases or skipped else 0) + PY + capture-linux: name: capture (linux, py${{ matrix.python-version }}) needs: scope @@ -1033,6 +1102,7 @@ jobs: - flow-lint - flow-unit - flow-critical + - flow-native-census - capture-linux - capture-macos - capture-windows diff --git a/llms-full.txt b/llms-full.txt index 92b612345..e1c66d62f 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -1342,6 +1342,38 @@ For an armed step: run's value into the recorded band and requires the whole band to match. A band that rests on a glyph-confusable identifier keeps the pixel comparison. +A verified identity says the row reads like the recorded row, not that no +other row reads the same. Before an armed step acts, the runtime counts the +rows that show the same identity (`openadapt_flow/runtime/identity_census.py`). When more +than one does, it acts only on the row with the record id recorded at teach +time, or halts. How each surface counts: + +- browser: every shown row in the page, with app-owned `data-*` ids and row + links as record ids; +- Windows (UI Automation agent): the sibling rows of the clicked row, with + non-positional AutomationIds as record ids. A grid that reports more rows + than it exposes, or a scrollable list that builds items on demand, cannot be + counted, and a row pick before a write halts; +- macOS (AX): the rows of the `AXTable`, `AXOutline` or `AXList` around the + element, with the AX identifier and URL as record ids. A table that reports + more `AXRows` than it exposes, or a list with no row count inside a scroll + area with a scroll bar, cannot be counted; +- Linux (AT-SPI): the sibling list items, table rows or tree items of the + clicked row, with AT-SPI accessible ids as record ids. A list with no Table + row count inside a scroll pane with a scroll bar cannot be counted. GTK 3 + list boxes and + tree views expose no row identity under the structured reading (one child + per row, or cells without a row), so the identity check there rests on OCR; +- pixel-only displays: the OCR rows of the frame, or, with no OCR band, every + place the identifier crop matches. Only a clear pixel mismatch rules a place + out; a frame at another scale than the recording cannot be counted. + +Ids that name a position (`row-3`, `r3c2`, `item_12`, the row's own index) +never tell rows apart. + +A surface that cannot count the rows halts any row pick that leads into a +write. + On a pixel-only display, the OCR tier also stops the correct record whenever its own identifier contains a look-alike glyph (`O`/`0`, `l`/`1`, `B`/`8`). A deployment can remove most of that cost by giving the runtime a complete diff --git a/packages/flow/docs/LIMITS.md b/packages/flow/docs/LIMITS.md index 4211a7e91..6baf20a49 100644 --- a/packages/flow/docs/LIMITS.md +++ b/packages/flow/docs/LIMITS.md @@ -185,6 +185,38 @@ For an armed step: run's value into the recorded band and requires the whole band to match. A band that rests on a glyph-confusable identifier keeps the pixel comparison. +A verified identity says the row reads like the recorded row, not that no +other row reads the same. Before an armed step acts, the runtime counts the +rows that show the same identity (`openadapt_flow/runtime/identity_census.py`). When more +than one does, it acts only on the row with the record id recorded at teach +time, or halts. How each surface counts: + +- browser: every shown row in the page, with app-owned `data-*` ids and row + links as record ids; +- Windows (UI Automation agent): the sibling rows of the clicked row, with + non-positional AutomationIds as record ids. A grid that reports more rows + than it exposes, or a scrollable list that builds items on demand, cannot be + counted, and a row pick before a write halts; +- macOS (AX): the rows of the `AXTable`, `AXOutline` or `AXList` around the + element, with the AX identifier and URL as record ids. A table that reports + more `AXRows` than it exposes, or a list with no row count inside a scroll + area with a scroll bar, cannot be counted; +- Linux (AT-SPI): the sibling list items, table rows or tree items of the + clicked row, with AT-SPI accessible ids as record ids. A list with no Table + row count inside a scroll pane with a scroll bar cannot be counted. GTK 3 + list boxes and + tree views expose no row identity under the structured reading (one child + per row, or cells without a row), so the identity check there rests on OCR; +- pixel-only displays: the OCR rows of the frame, or, with no OCR band, every + place the identifier crop matches. Only a clear pixel mismatch rules a place + out; a frame at another scale than the recording cannot be counted. + +Ids that name a position (`row-3`, `r3c2`, `item_12`, the row's own index) +never tell rows apart. + +A surface that cannot count the rows halts any row pick that leads into a +write. + On a pixel-only display, the OCR tier also stops the correct record whenever its own identifier contains a look-alike glyph (`O`/`0`, `l`/`1`, `B`/`8`). A deployment can remove most of that cost by giving the runtime a complete diff --git a/packages/flow/openadapt_flow/backends/linux_backend.py b/packages/flow/openadapt_flow/backends/linux_backend.py index fcebd00a7..f79aa24bc 100644 --- a/packages/flow/openadapt_flow/backends/linux_backend.py +++ b/packages/flow/openadapt_flow/backends/linux_backend.py @@ -585,6 +585,25 @@ def locate_structural( supported_operations=list(candidate.supported_operations), ) + def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the row at (x, y). + + See :class:`openadapt_flow.backend.IdentityCandidatesBackend`. Uses the + client's optional ``row_census``; a client without it cannot count + rows (None). Offsets are global pixels, the same scale as clicks. + """ + census_at = getattr(self._client, "row_census", None) + if not callable(census_at): + return None + try: + window, global_x, global_y = self._global_point( + int(x), int(y), require_active=False + ) + census = census_at(window, global_x, global_y, limit=2000) + except Exception: + return None + return census if isinstance(census, dict) else None + def displayed_text_at(self, x: int, y: int) -> Optional[str]: """The text of the one AT-SPI element under (x, y), for a read-back. @@ -1358,12 +1377,7 @@ def structured_text(self, element: LinuxElement) -> Optional[str]: own = element.native node = own row = None - row_roles = { - "list item", - "table cell", - "table row", - "tree item", - } + row_roles = self._ROW_ROLES for _ in range(8): if self._raw_role(node) in row_roles: row = node @@ -1384,6 +1398,13 @@ def structured_text(self, element: LinuxElement) -> Optional[str]: break branch = parent + return self._row_text(row, branch) + + _ROW_ROLES = frozenset({"list item", "table cell", "table row", "tree item"}) + + def _row_text(self, row: Any, branch: Any) -> Optional[str]: + """A row's identity: its children's text except the target's branch.""" + parts: list[str] = [] for child in self._children(row): if child is branch: @@ -1403,6 +1424,45 @@ def structured_text(self, element: LinuxElement) -> Optional[str]: parts.append(text) return _clean_text(" ".join(parts)) + def row_census( + self, window: LinuxWindow, x: int, y: int, *, limit: int + ) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the row at (x, y). + + Mirrors :meth:`structured_text`: the record row is the nearest list + item, table row or cell, tree item; its identity is its children's + text except the target's branch; every sibling row is read the same + way. Record ids are AT-SPI accessible ids on the row and the target's + branch. A table whose Table interface reports more rows than it + exposes, or a row whose children are not built, is ``incomplete``. + Offsets are in global screen pixels. None on any failure. + """ + try: + from openadapt_flow.runtime.identity_census import tree_row_census + + element = self.element_at_point(window, x, y) + if element is None: + return None + client = self + + def identity(row: Any, path: list[int]) -> Optional[str]: + if not path: + return None + children = client._children(row) + if path[0] >= len(children): + return None + return client._row_text(row, children[path[0]]) + + return tree_row_census( + element.native, + _AtspiTreeView(self), + row_roles=self._ROW_ROLES, + identity=identity, + max_rows=limit, + ) + except Exception: + return None + def perform_native(self, element: LinuxElement, operation: str) -> str: if operation == "focus": component = self._call( @@ -1526,3 +1586,54 @@ def physical_scroll(self, dx: int, dy: int) -> bool: # not provide a reliable pixel-delta contract. Refuse rather than # fabricate delivery evidence. return False + + +class _AtspiTreeView: + """``identity_census.TreeView`` over live AT-SPI nodes (screen pixels).""" + + def __init__(self, client: AtspiLinuxClient) -> None: + self._client = client + + def parent(self, node: Any) -> Any: + return self._client._call(node, "get_parent", default=None) + + def children(self, node: Any) -> list[Any]: + return self._client._children(node) + + def role(self, node: Any) -> Optional[str]: + return self._client._raw_role(node) or None + + def origin(self, node: Any) -> Optional[tuple[float, float]]: + bounds = self._client._bounds(node) + return (float(bounds[0]), float(bounds[1])) if bounds is not None else None + + def id_attributes(self, node: Any) -> list[tuple[str, str]]: + value = self._client._accessible_id(node) + return [("atspi-id", value)] if value else [] + + def row_count(self, container: Any) -> Optional[int]: + table = self._client._call( + container, "get_table_iface", "queryTable", default=None + ) + if table is None: + return None + count = self._client._call(table, "get_n_rows", default=None) + if count is None: + count = getattr(table, "nRows", None) + try: + return int(count) if count is not None else None + except (TypeError, ValueError): + return None + + def scrollable(self, container: Any) -> Optional[bool]: + """True when an enclosing scroll pane holds a scroll bar.""" + node = container + for _ in range(3): + if node is None: + return False + if self.role(node) == "scroll pane": + return any( + self.role(child) == "scroll bar" for child in self.children(node) + ) + node = self.parent(node) + return False diff --git a/packages/flow/openadapt_flow/backends/macos_backend.py b/packages/flow/openadapt_flow/backends/macos_backend.py index 48f031175..e52bc9372 100644 --- a/packages/flow/openadapt_flow/backends/macos_backend.py +++ b/packages/flow/openadapt_flow/backends/macos_backend.py @@ -272,6 +272,14 @@ def structured_text(self, element: MacElement) -> Optional[str]: """Exact AX text of ``element`` (value/title/description), or None.""" ... + # Optional (IdentityCandidatesBackend): a client without it cannot count + # look-alike rows, and the runtime halts a row pick before a write. + # + # def row_census(self, pid: int, window_title: str, x: float, y: float, *, + # limit: int) -> Optional[dict[str, Any]]: + # """identity_candidates_at's census at screen point (x, y), with row + # offsets in screen points.""" + def _clean_text(value: object) -> Optional[str]: """Collapse whitespace to a single-spaced string, or None if empty.""" @@ -529,6 +537,140 @@ def structured_text(self, element: MacElement) -> Optional[str]: except Exception: # noqa: BLE001 - a momentary AX failure is not fatal return None + def row_census( + self, pid: int, window_title: str, x: float, y: float, *, limit: int + ) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the element at (x, y). + + The element's identity is its own AX text (as ``structured_text``). + Its record row is the enclosing ``AXRow``, or the item of an + ``AXList``; every sibling row is read at the same child path. Record + ids are the AX identifier and URL of the row and of the target's + node. A table that reports more ``AXRows`` than it exposes as + children, or a row whose cells are not built, is ``incomplete``. + Offsets are in screen points. None on any AX failure. + """ + try: + from ApplicationServices import ( + AXUIElementCopyAttributeValue, + AXUIElementCopyElementAtPosition, + AXUIElementCreateApplication, + kAXTitleAttribute, + kAXTopLevelUIElementAttribute, + ) + + from openadapt_flow.runtime.identity_census import tree_row_census + + app = AXUIElementCreateApplication(int(pid)) + hit_error, element = AXUIElementCopyElementAtPosition( + app, float(x), float(y), None + ) + if hit_error != 0 or element is None: + return None + top_error, top_level = AXUIElementCopyAttributeValue( + element, kAXTopLevelUIElementAttribute, None + ) + if top_error != 0 or top_level is None: + return None + title_error, title = AXUIElementCopyAttributeValue( + top_level, kAXTitleAttribute, None + ) + if title_error != 0 or str(title or "") != window_title: + return None + + def identity(row: Any, path: list[int]) -> Optional[str]: + node = row + for index in path: + children = _AXTreeView(self).children(node) + if index >= len(children): + return None + node = children[index] + return _clean_text(self._text(node)) + + return tree_row_census( + element, + _AXTreeView(self), + row_roles=frozenset({"AXRow"}), + container_roles=frozenset({"AXList"}), + identity=identity, + max_rows=limit, + ) + except Exception: # noqa: BLE001 - unobservable AX is an ordinary miss + return None + + +class _AXTreeView: + """``identity_census.TreeView`` over live AX elements (screen points).""" + + def __init__(self, client: QuartzMacAXClient) -> None: + self._client = client + + @staticmethod + def same(a: Any, b: Any) -> bool: + # Every AX read returns a new wrapper; CFEqual compares the elements. + from CoreFoundation import CFEqual + + return bool(CFEqual(a, b)) + + def parent(self, node: Any) -> Any: + from ApplicationServices import kAXParentAttribute + + return self._client._attr(node, kAXParentAttribute) + + def children(self, node: Any) -> list[Any]: + from ApplicationServices import kAXChildrenAttribute + + return list(self._client._attr(node, kAXChildrenAttribute) or []) + + def role(self, node: Any) -> Optional[str]: + from ApplicationServices import kAXRoleAttribute + + raw = self._client._attr(node, kAXRoleAttribute) + return str(raw) if raw is not None else None + + def origin(self, node: Any) -> Optional[tuple[float, float]]: + bounds = self._client._bounds(node) + return (bounds[0], bounds[1]) if bounds is not None else None + + def id_attributes(self, node: Any) -> list[tuple[str, str]]: + from ApplicationServices import kAXIdentifierAttribute, kAXURLAttribute + + found = [] + for name, attribute in ( + ("ax-identifier", kAXIdentifierAttribute), + ("ax-url", kAXURLAttribute), + ): + value = self._client._attr(node, attribute) + if value is not None and str(value).strip(): + found.append((name, str(value))) + return found + + def row_count(self, container: Any) -> Optional[int]: + from ApplicationServices import kAXRowsAttribute + + rows = self._client._attr(container, kAXRowsAttribute) + if rows is None: + return None + try: + return len(rows) + except TypeError: + return None + + def scrollable(self, container: Any) -> Optional[bool]: + """True when an enclosing AXScrollArea has a vertical scroll bar.""" + from ApplicationServices import kAXVerticalScrollBarAttribute + + node = container + for _ in range(3): + if node is None: + return False + if self.role(node) == "AXScrollArea": + return ( + self._client._attr(node, kAXVerticalScrollBarAttribute) is not None + ) + node = self.parent(node) + return False + _NATIVE_MODIFIER_FLAGS = { "ctrl": "control", @@ -1222,6 +1364,40 @@ def locate_structural( supported_operations=list(candidate.supported_operations), ) + def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the element at (x, y). + + See :class:`openadapt_flow.backend.IdentityCandidatesBackend`. Uses the + AX client's optional ``row_census``; a client without it cannot count + rows (None). Row offsets are converted to captured pixels. + """ + census_at = getattr(self._ax_client, "row_census", None) + if not callable(census_at): + return None + try: + window = self._ensure_captured() + screen_x, screen_y = self._pixel_to_screen(window, int(x), int(y)) + census = census_at( + window.pid, + window.title, + screen_x, + screen_y, + limit=_AX_ENUMERATION_LIMIT, + ) + except Exception: # noqa: BLE001 - a momentary AX failure is not fatal + return None + if not isinstance(census, dict): + return None + scale_x = self._scale_x or 1.0 + scale_y = self._scale_y or 1.0 + for candidate in census.get("candidates") or []: + dx, dy = candidate.get("dx"), candidate.get("dy") + if isinstance(dx, (int, float)) and isinstance(dy, (int, float)): + candidate["dx"], candidate["dy"] = dx * scale_x, dy * scale_y + else: + candidate["dx"] = candidate["dy"] = None + return census + def displayed_text_at(self, x: int, y: int) -> Optional[str]: """The text of the one AX element under (x, y), for a read-back. diff --git a/packages/flow/openadapt_flow/backends/playwright_backend.py b/packages/flow/openadapt_flow/backends/playwright_backend.py index 57eef2c85..094d97597 100644 --- a/packages/flow/openadapt_flow/backends/playwright_backend.py +++ b/packages/flow/openadapt_flow/backends/playwright_backend.py @@ -711,7 +711,18 @@ class _FramePoint: const POSITIONAL = /(?:index|idx|position|pos|order|rank|seq)/i; const parent = row.parentElement; const index = parent ? Array.prototype.indexOf.call(parent.children, row) : -1; + // Mirror of identity_census.positional_id: an id that names a position + // (row-3, r3c2, item_12, the row's own index) moves with the row. + const WORDS = 'row|item|idx|index|cell|col|column|line|pos|position|listviewitem|node|child'; + const POSITION_TOKEN = new RegExp('^(?:(?:' + WORDS + ')\\d+|r\\d+c\\d+)$', 'i'); + const POSITION_WORD = new RegExp('^(?:' + WORDS + '|r|c)$', 'i'); const positional = (value) => { + const tokens = value.split(/[^A-Za-z0-9]+/).filter(Boolean); + if (tokens.some((t) => POSITION_TOKEN.test(t))) return true; + for (let i = 0; i + 1 < tokens.length; i += 1) { + if (POSITION_WORD.test(tokens[i]) && /^\d+$/.test(tokens[i + 1])) return true; + } + if (tokens.length > 1 && tokens.every((t) => /^\d{1,3}$/.test(t))) return true; const m = /^\D*(\d+)\D*$/.exec(value); if (!m) return false; const n = Number(m[1]); diff --git a/packages/flow/openadapt_flow/backends/win_agent/server.py b/packages/flow/openadapt_flow/backends/win_agent/server.py index a56b996a0..1ba38b2aa 100644 --- a/packages/flow/openadapt_flow/backends/win_agent/server.py +++ b/packages/flow/openadapt_flow/backends/win_agent/server.py @@ -12,6 +12,7 @@ serialized request POST /uia/locator-at -> stable UIA locator at a demonstrated point POST /uia/text-at-point -> structured row text for identity verification + POST /uia/identity-candidates-at -> rows that read the same as that row POST /uia/find -> zero / unique / ambiguous exact candidates POST /uia/act -> unique native UIA action + delivery receipt POST /uia/read -> read-only text of one unique control (value or @@ -82,6 +83,7 @@ import io import json import os +import re import secrets import subprocess import sys @@ -734,6 +736,208 @@ def _structured_text_at(auto: Any, x: int, y: int) -> Optional[str]: return value or None +_CENSUS_ROW_TYPES = frozenset( + {"DataItemControl", "ListItemControl", "TreeItemControl", "TableRowControl"} +) +_MAX_CENSUS_ROWS = 2000 +_SINGLE_NUMBER = re.compile(r"^\D*(\d+)\D*$") +# Mirror of identity_census.positional_id. +_POSITION_WORDS = ( + "row|item|idx|index|cell|col|column|line|pos|position|listviewitem|node|child" +) +_POSITION_TOKEN = re.compile(rf"^(?:(?:{_POSITION_WORDS})\d+|r\d+c\d+)$", re.IGNORECASE) +_POSITION_WORD_TOKEN = re.compile(rf"^(?:{_POSITION_WORDS}|r|c)$", re.IGNORECASE) + + +def _positional_id(value: str, index: int) -> bool: + """Whether a row id names a position (``row-3``, ``r3c2``) not a record.""" + + tokens = [t for t in re.split(r"[^A-Za-z0-9]+", value) if t] + if any(_POSITION_TOKEN.match(t) for t in tokens): + return True + for word, following in zip(tokens, tokens[1:]): + if _POSITION_WORD_TOKEN.match(word) and following.isdigit(): + return True + if len(tokens) > 1 and all(t.isdigit() and len(t) <= 3 for t in tokens): + return True + match = _SINGLE_NUMBER.match(value) + return bool(match and int(match.group(1)) in (index, index + 1)) + + +def _row_identity(row: object, excluded: Optional[int]) -> Optional[str]: + """A row's identity as ``_structured_text_at`` reads it, by child index.""" + + parts: list[str] = [] + for index, child in enumerate(_children(row)): + if excluded is not None and index == excluded: + continue + name = str(_control_value(child, "Name", "") or "") + if name: + parts.append(name) + if not parts: + name = str(_control_value(row, "Name", "") or "") + if name: + parts.append(name) + value = " ".join(" ".join(parts).split()) + return value or None + + +def _census_record_key( + row: object, branch: Optional[object], index: int +) -> Optional[str]: + """The row's AutomationIds (row and target branch), minus positional ones.""" + + keys: set[str] = set() + for control in (row, branch): + if control is None: + continue + value = str(_control_value(control, "AutomationId", "") or "").strip() + if not value or len(value) > 256: + continue + if _positional_id(value, index): + continue + keys.add(f"automation-id={value}") + return "\n".join(sorted(keys)) or None + + +def _identity_candidates_at(auto: Any, x: int, y: int) -> Optional[dict[str, Any]]: + """Census of the shown rows that read the same as the row at (x, y). + + Mirrors ``_structured_text_at`` (and ``identity_census.tree_row_census`` + in the openadapt_flow package, which this guest copy cannot import): the + record row is the nearest data, list, tree or table-row item; its identity + is its children's names except the clicked cell; every sibling row of the + same control type is read at the same child index. A container whose Grid + pattern reports more rows than it exposes, or a scrollable container that + realizes items on demand (ItemContainer pattern), is ``incomplete``: its + off-screen rows are not in the tree. ``{"row": False}`` when the point is + in no row; None when UIA fails. + """ + try: + element = auto.ControlFromPoint(x, y) + except Exception: # noqa: BLE001 + return None + if element is None: + return None + row: Optional[object] = element + for _ in range(6): + if str(_control_value(row, "ControlTypeName", "")) in _CENSUS_ROW_TYPES: + break + row = _parent(row) + if row is None: + break + else: + row = None + if row is None or ( + str(_control_value(row, "ControlTypeName", "")) not in _CENSUS_ROW_TYPES + ): + return {"row": False} + own: Optional[object] = element + for _ in range(6): + parent = _parent(own) + if parent is None: + own = None + break + if parent is row: + break + own = parent + row_children = _children(row) + excluded = next( + (i for i, child in enumerate(row_children) if own is not None and child is own), + None, + ) + branch = row_children[excluded] if excluded is not None else None + target_identity = _row_identity(row, excluded) + if not target_identity: + return None + container = _parent(row) + row_type = str(_control_value(row, "ControlTypeName", "")) + siblings = _children(container) if container is not None else [row] + rows = [ + c + for c in siblings + if c is row or str(_control_value(c, "ControlTypeName", "")) == row_type + ] + + # The target row's own wrapper may differ from the sibling's; locate it by + # runtime id when the provider exposes one, else by bounds. + def same(a: object, b: object) -> bool: + if a is b: + return True + compare = getattr(auto, "ControlsAreSame", None) + if callable(compare): + try: + return bool(compare(a, b)) + except Exception: # noqa: BLE001 + pass + return _bounds(a) is not None and _bounds(a) == _bounds(b) + + row_index = next((i for i, c in enumerate(rows) if same(c, row)), None) + incomplete = len(rows) > _MAX_CENSUS_ROWS or row_index is None + if container is not None and not incomplete: + # A provider that fails while answering is not a complete answer. + try: + grid_get = _control_value(container, "GetGridPattern", None) + grid = grid_get() if callable(grid_get) else None + declared = int(grid.RowCount) if grid is not None else None + scroll_get = _control_value(container, "GetScrollPattern", None) + scroll = scroll_get() if callable(scroll_get) else None + scrollable = ( + bool(scroll.VerticallyScrollable) if scroll is not None else False + ) + items_get = _control_value(container, "GetItemContainerPattern", None) + realizes = callable(items_get) and items_get() is not None + except Exception: # noqa: BLE001 + declared, scrollable, realizes, incomplete = None, True, True, True + if declared is not None and declared > len(rows): + incomplete = True + if scrollable and realizes: + incomplete = True + result: dict[str, Any] = { + "record_key": _census_record_key(row, branch, row_index or 0), + "incomplete": incomplete, + "top_level": True, + "candidates": [], + } + if incomplete: + return result + origin = _bounds(row) + for index, candidate in enumerate(rows): + is_self = index == row_index + if is_self: + observed: Optional[str] = target_identity + candidate_branch = branch + else: + observed = _row_identity(candidate, excluded) + kids = _children(candidate) + if row_children and not kids: + # A row whose cells are not built yet: it cannot be read. + result["incomplete"] = True + result["candidates"] = [] + return result + candidate_branch = ( + kids[excluded] + if excluded is not None and excluded < len(kids) + else None + ) + if observed != target_identity: + continue + here = _bounds(candidate) + result["candidates"].append( + { + "self": is_self, + "record_key": ( + result["record_key"] + if is_self + else _census_record_key(candidate, candidate_branch, index) + ), + "dx": here[0] - origin[0] if here and origin else None, + "dy": here[1] - origin[1] if here and origin else None, + } + ) + return result + + _ROW_TYPES = frozenset( { "DataItemControl", @@ -844,12 +1048,19 @@ def _perform_uia_initialized( auto: Any, operation: str, payload: dict[str, Any] ) -> dict[str, Any]: """Perform one typed UIA operation in an initialized COM apartment.""" - if operation in {"locator-at", "text-at-point", "focused-at-point"}: + if operation in { + "locator-at", + "text-at-point", + "focused-at-point", + "identity-candidates-at", + }: data = _exact_object(payload, required=frozenset({"x", "y"}), label=operation) x = _bounded_int(data["x"], "x") y = _bounded_int(data["y"], "y") if operation == "text-at-point": return {"status": "ok", "text": _structured_text_at(auto, x, y)} + if operation == "identity-candidates-at": + return {"status": "ok", "census": _identity_candidates_at(auto, x, y)} try: element = auto.ControlFromPoint(x, y) except Exception as exc: # noqa: BLE001 @@ -1432,6 +1643,7 @@ def do_POST(self) -> None: # noqa: N802 - stdlib naming "/context/identity": ("context", "identity"), "/uia/locator-at": ("uia", "locator-at"), "/uia/text-at-point": ("uia", "text-at-point"), + "/uia/identity-candidates-at": ("uia", "identity-candidates-at"), "/uia/find": ("uia", "find"), "/uia/act": ("uia", "act"), "/uia/read": ("uia", "read"), diff --git a/packages/flow/openadapt_flow/backends/windows_backend.py b/packages/flow/openadapt_flow/backends/windows_backend.py index ecb96191d..ecbc51834 100644 --- a/packages/flow/openadapt_flow/backends/windows_backend.py +++ b/packages/flow/openadapt_flow/backends/windows_backend.py @@ -40,7 +40,7 @@ import struct import warnings from dataclasses import dataclass -from typing import Callable, Optional +from typing import Any, Callable, Optional from urllib.parse import urlparse import requests @@ -687,6 +687,22 @@ def structured_text_at(self, x: int, y: int) -> Optional[str]: return None return str(value) + def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the row under (x, y). + + See :class:`openadapt_flow.backend.IdentityCandidatesBackend`. Served by + the agent's typed ``/uia/identity-candidates-at``; an agent without it, + or the legacy exec path, cannot count rows (None), and the runtime + halts a row pick before a write. + """ + typed = self._post_typed_read( + "/uia/identity-candidates-at", {"x": int(x), "y": int(y)} + ) + if not typed.available or typed.payload is None: + return None + census = typed.payload.get("census") + return census if isinstance(census, dict) else None + def _execute_read(self, command: str) -> Optional[str]: """POST bare Python to WAA and return the response body if any. diff --git a/packages/flow/openadapt_flow/risk.py b/packages/flow/openadapt_flow/risk.py index 8a61623f1..687d5c8b0 100644 --- a/packages/flow/openadapt_flow/risk.py +++ b/packages/flow/openadapt_flow/risk.py @@ -300,8 +300,9 @@ def string_tuple(field: str) -> tuple[str, ...]: # "postal". Ordered roughly by how common they are on real controls. # # "sign" counts in its writing sense ("Sign visit note", "Co-sign", "Signed"), -# the clinical attestation that locks a note, but never as "sign in", "sign -# on" or "sign out", which are session chrome. +# the clinical attestation that locks a note, but never as "sign in" or "sign +# on", and "sign out" / "log out" only with a clinical object ("Sign out +# patient"), never as the session's own "Sign out". # # Deliberately EXCLUDED to avoid noisy false positives on navigation chrome: # bare "post"/"order"/"book"/"complete" (collide with "Posts"/"Orders"/ @@ -323,6 +324,14 @@ def string_tuple(field: str) -> tuple[str, ...]: r"sign[\s\-_]?up", r"signup", r"(?:co[\-_]?|e[\-_]?)?sign(?:s|ed|ing)?(?![\s\-_]*(?:in|on|out)\b)", + # "Sign out patient", "Log out Jane Doe": signing out a record (a + # discharge or check-out) is a write. Only a bare "sign out" / "log out", + # or one that names the session ("sign out of OpenAdapt", "sign out now"), + # is session chrome. + r"(?:sign|log)(?:s|ged|ging|ed|ing)?[\s\-_]*out[\s\-_]+" + r"(?!(?:of|now|here|and|from|all|everywhere|completely|securely|safely)\b)" + r"[a-z]+", + r"discharg(?:e|es|ed|ing)", r"regist(?:er|ers|ering|ration)", r"enroll(?:s|ed|ing|ment)?", r"add(?:s|ed|ing)?", @@ -605,7 +614,9 @@ def step_text(step: Step) -> str: drag_end_structural = drag_end.structural if drag_end is not None else None if drag_end_structural is not None and drag_end_structural.name: parts.append(drag_end_structural.name) - return " ".join(parts) + # " | " keeps a write phrase from spanning two labels: "Sign out" twice + # must not read as "sign out Sign". + return " | ".join(parts) def infer_step_risk( diff --git a/packages/flow/openadapt_flow/runtime/identity_census.py b/packages/flow/openadapt_flow/runtime/identity_census.py index dc7a7d76d..4164b7716 100644 --- a/packages/flow/openadapt_flow/runtime/identity_census.py +++ b/packages/flow/openadapt_flow/runtime/identity_census.py @@ -33,7 +33,8 @@ from __future__ import annotations -from typing import Any, Callable, Literal, Mapping, NamedTuple, Optional +import re +from typing import Any, Callable, Literal, Mapping, NamedTuple, Optional, Protocol CensusVerdict = Literal[ "unique", @@ -140,6 +141,256 @@ def count_matching_rows( return others + 1 +# --------------------------------------------------------------------------- +# Row census over an accessibility tree (macOS AX, Linux AT-SPI) +# --------------------------------------------------------------------------- +# +# The browser backend counts look-alike rows in page JavaScript. A native +# accessibility tree is walked here instead, through a small adapter, so the +# rules are the same on every backend: find the record row that holds the +# target, read the same identity from every sibling row, and report each row +# that reads the same with its stable record id. The Windows agent runs in the +# guest without this package and carries a mirror of this function +# (backends/win_agent/server.py, ``_identity_candidates_at``). + +#: Rows read before a census gives up and reports ``incomplete``. +TREE_CENSUS_ROW_CAP = 2000 + +_SINGLE_NUMBER = re.compile(r"^\D*(\d+)\D*$") +_POSITION_WORDS = ( + "row|item|idx|index|cell|col|column|line|pos|position|listviewitem|node|child" +) +# A whole token that is a position word with its number (row3, item_12 is +# split into item + 12) or a grid coordinate (r3c2). +_POSITION_TOKEN = re.compile(rf"^(?:(?:{_POSITION_WORDS})\d+|r\d+c\d+)$", re.IGNORECASE) +_POSITION_WORD_TOKEN = re.compile(rf"^(?:{_POSITION_WORDS}|r|c)$", re.IGNORECASE) + + +def positional_id(value: str, index: int) -> bool: + """Whether a row id names a position rather than a record. + + Such an id moves with the row (``row-3``, ``r3c2``, ``item_12``, a bare + ``3``): after the rows reorder or scroll it points at whichever record now + sits there, so it never tells rows apart. The id is split into + alphanumeric tokens. It is positional when a token is a position word with + its number (``row3``, ``r3c2``), when a position word token is followed by + a number token (``row-3``, ``item_12``, ``row-3-col-1``), when every token + is a short number (grid coordinates such as ``3-1``), or when its only + number is the row's index (0- or 1-based). UUIDs, ULIDs, MRNs and + composite record ids (``ENC2024-000123``) are not. + """ + tokens = [t for t in re.split(r"[^A-Za-z0-9]+", value) if t] + if any(_POSITION_TOKEN.match(t) for t in tokens): + return True + for word, following in zip(tokens, tokens[1:]): + if _POSITION_WORD_TOKEN.match(word) and following.isdigit(): + return True + if len(tokens) > 1 and all(t.isdigit() and len(t) <= 3 for t in tokens): + return True + match = _SINGLE_NUMBER.match(value) + return bool(match and int(match.group(1)) in (index, index + 1)) + + +class TreeView(Protocol): + """Read-only access to one accessibility tree, for :func:`tree_row_census`. + + Nodes are opaque. ``origin`` is the node's top-left in the backend's click + coordinates, or None when it cannot be mapped. ``id_attributes`` lists the + node's candidate record ids as ``(name, value)`` pairs (an AX identifier, + an AT-SPI ``id`` attribute, a URL). ``row_count`` is the number of rows the + container says it has, when it says (None otherwise). ``scrollable`` says + whether a container without a row count can scroll rows out of view; + anything but a definite False makes the census ``incomplete``. + """ + + def parent(self, node: Any) -> Any: ... + + def children(self, node: Any) -> list[Any]: ... + + def role(self, node: Any) -> Optional[str]: ... + + def origin(self, node: Any) -> Optional[tuple[float, float]]: ... + + def id_attributes(self, node: Any) -> list[tuple[str, str]]: ... + + def row_count(self, container: Any) -> Optional[int]: ... + + def scrollable(self, container: Any) -> Optional[bool]: + """Whether the container can scroll rows out of view; None unknown.""" + ... + + +def _same(view: TreeView, a: Any, b: Any) -> bool: + """Whether two node handles are the same node. + + A view MAY define ``same(a, b)``: AX hands back a fresh wrapper for the + same element on every read, so equality there is ``CFEqual``. + """ + if a is b: + return True + same = getattr(view, "same", None) + if callable(same): + try: + return bool(same(a, b)) + except Exception: + return False + return bool(a == b) + + +def _record_key(view: TreeView, row: Any, path: list[int], index: int) -> Optional[str]: + """The stable record id of ``row``: ids on the row and on the path to the + target, minus values that are only the row's position.""" + + parts: set[str] = set() + node = row + nodes = [row] + for step in path: + children = view.children(node) + if step >= len(children): + break + node = children[step] + nodes.append(node) + for item in nodes: + for name, value in view.id_attributes(item): + value = (value or "").strip() + if not value or len(value) > 256: + continue + if positional_id(value, index): + continue + parts.add(f"{name}={value}") + return "\n".join(sorted(parts)) or None + + +def tree_row_census( + target: Any, + view: TreeView, + *, + row_roles: frozenset[str], + identity: Callable[[Any, list[int]], Optional[str]], + container_roles: frozenset[str] = frozenset(), + max_depth: int = 8, + max_rows: int = TREE_CENSUS_ROW_CAP, +) -> Optional[dict[str, Any]]: + """Count the sibling rows that read the same identity as ``target``'s row. + + Args: + target: The node under the point. + view: The tree adapter. + row_roles: Roles that make a node a record row. + identity: ``identity(row, path)`` reads a row's identity the way the + backend's ``structured_text_at`` does, where ``path`` is the child + index path from the row to the target's place in it. None when the + row has nothing to read there. + container_roles: Roles of a list or table whose direct children are + its rows even when they carry no row role (an ``AXList`` item). + max_depth: How far above ``target`` to look for its row. + max_rows: More sibling rows than this reports ``incomplete``. + + Returns: + The census shape of ``IdentityCandidatesBackend.identity_candidates_at``, + ``{"row": False}`` when the target is in no row, or None when the + target's own identity cannot be read. + """ + row = None + previous = None + node = target + for _ in range(max_depth + 1): + if node is None: + break + role = view.role(node) + if role in row_roles: + row = node + break + if role in container_roles and previous is not None: + row = previous + break + previous, node = node, view.parent(node) + if row is None: + return {"row": False} + path: list[int] = [] + node = target + while not _same(view, node, row): + parent = view.parent(node) + if parent is None: + return None + siblings = view.children(parent) + try: + path.insert( + 0, next(i for i, c in enumerate(siblings) if _same(view, c, node)) + ) + except StopIteration: + return None + node = parent + own_identity = identity(row, path) + if not own_identity: + return None + container = view.parent(row) + siblings = view.children(container) if container is not None else [row] + row_role = view.role(row) + rows = [c for c in siblings if _same(view, c, row) or view.role(c) == row_role] + declared = view.row_count(container) if container is not None else None + if declared is not None: + hidden_rows = declared > len(rows) + elif container is None: + hidden_rows = False + else: + # No row count to check against: only a container that cannot + # scroll shows every row it has. Unknown scrollability is incomplete. + try: + hidden_rows = view.scrollable(container) is not False + except Exception: + hidden_rows = True + incomplete = len(rows) > max_rows or hidden_rows + row_index = next((i for i, c in enumerate(rows) if _same(view, c, row)), 0) + result: dict[str, Any] = { + "record_key": _record_key(view, row, path, row_index), + "incomplete": incomplete, + "top_level": True, + "candidates": [], + } + if incomplete: + return result + origin = view.origin(row) + for index, candidate in enumerate(rows): + is_row = _same(view, candidate, row) + if is_row: + observed: Optional[str] = own_identity + else: + node = candidate + for step in path: + children = view.children(node) + if step >= len(children): + # The row is not populated where the target sits (a lazily + # built row): its identity cannot be read, so not every + # row was counted. + result["incomplete"] = True + result["candidates"] = [] + return result + node = children[step] + observed = identity(candidate, path) + if observed != own_identity: + continue + here = view.origin(candidate) + offset = ( + (here[0] - origin[0], here[1] - origin[1]) + if here is not None and origin is not None + else (None, None) + ) + result["candidates"].append( + { + "self": is_row, + "record_key": ( + result["record_key"] + if is_row + else _record_key(view, candidate, path, index) + ), + "dx": offset[0], + "dy": offset[1], + } + ) + return result + + # Structural roles (backend-neutral, as the recorder stores them) of a form # field or a control. A target recorded with one of these is not a record row. CONTROL_ROLES = frozenset( diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 389316b0d..5aae38dde 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -736,6 +736,16 @@ class _WebFormApprovalView: ref: Any +# The pixel census's OCR tie-break: a place the pixels do not rule out is +# distinguishable only when OCR reads both crops with every line at or above +# this confidence, neither text contains the other, and the two texts are less +# similar than this ratio. One label read two ways measured 0.82-0.95 +# ("Smith, John" / "Smith, S John" / "Smith, 10 lohn"); sibling tab labels +# ("Contact" / "Stats") measure 0.5 or less. +PIXEL_CENSUS_OCR_MIN_CONFIDENCE = 0.8 +PIXEL_CENSUS_OCR_MAX_SIMILARITY = 0.6 + + class Replayer: """Replays a Workflow against a Backend using injected vision. @@ -15571,7 +15581,7 @@ def ocr_tier() -> Optional[IdentityCheck]: [structured_tier, pixel_tier, vlm_tier, ocr_tier] ) return self._apply_candidate_census( - step, resolution, check, before_png, params, workflow + step, resolution, check, before_png, params, workflow, bundle_dir ) def _structured_census(self, step: Step, point: Point) -> census_mod.CensusDecision: @@ -15690,6 +15700,7 @@ def _apply_candidate_census( before_png: bytes, params: Mapping[str, RuntimeParamScalar], workflow: Workflow, + bundle_dir: Optional[Path] = None, ) -> IdentityCheck: """Keep ``verified`` only when the identity picks one shown record. @@ -15712,6 +15723,12 @@ def _apply_candidate_census( decision = self._structured_census(step, resolution.point) else: decision = self._ocr_census(step, resolution, before_png, params, workflow) + if decision.verdict == "cannot_count" and check.mode in ("pixel", "vlm"): + # No OCR band to read: count the places the identifier crop + # itself verifies instead. + decision = self._pixel_census( + step, resolution, before_png, workflow, bundle_dir + ) if decision.verdict == "cannot_count" and teach_role == "control": # Teach time recorded a form field or a control, and no census can # count here. A census that can count always decides. @@ -15749,6 +15766,121 @@ def _apply_candidate_census( } ) + def _pixel_census( + self, + step: Step, + resolution: Resolution, + before_png: bytes, + workflow: Workflow, + bundle_dir: Optional[Path], + ) -> census_mod.CensusDecision: + """Count the places on the frame that could show the identifier crop. + + For a pixel identity with no OCR band (a remote display): every place + the recorded identifier crop matches is cut from the frame and checked + with the ladder's pixel verifier. A clear ``mismatch`` rules a place + out, and so does its own text when OCR reads both crops and the text + differs (sibling tabs share their chrome, not their label). A verify, + an abstain (drift, the uncertain band, verify turned off) or no verdict + with the same or unreadable text counts it as a look-alike. The census is + ``uncounted`` when the frame has too many places, when a place cannot + be cut, or when the target's own place is not among them (the frame + renders at another scale than the recorded crop). A featureless crop + (a blank band, a ruled line) cannot count at all: ``cannot_count``. + """ + from openadapt_flow.vision.match import ( + AMBIGUITY_SUSPICION_SCORE, + crop_is_featureless, + template_peaks, + ) + + anchor = step.anchor + assert anchor is not None + recorded, live = self._identifier_crops( + anchor, resolution, before_png, bundle_dir, workflow=workflow + ) + if recorded is None or live is None or anchor.identifier_region is None: + return census_mod.CensusDecision("cannot_count") + try: + if crop_is_featureless(recorded): + # A blank band or a ruled line matches everywhere along it: it + # cannot count rows (cannot_count, not "ambiguous"). + return census_mod.CensusDecision("cannot_count") + peaks = template_peaks( + before_png, recorded, threshold=AMBIGUITY_SUSPICION_SCORE + ) + except Exception: + return census_mod.CensusDecision("cannot_count") + if peaks is None: + return census_mod.CensusDecision("uncounted") + rx, ry, rw, rh = anchor.identifier_region + dx, dy = self._identifier_displacement( + anchor, resolution, before_png, bundle_dir, workflow=workflow + ) + target = (rx + dx, ry + dy) + + def is_target(x: int, y: int) -> bool: + return abs(x - target[0]) < max(1, rw // 2) and abs(y - target[1]) < max( + 1, rh // 2 + ) + + if not any(is_target(x, y) for x, y in peaks): + return census_mod.CensusDecision("uncounted") + + def crop_text(png: bytes) -> str: + """The crop's own text, squashed; empty when OCR reads nothing.""" + try: + lines = self.vision.ocr(png, region=None) + except Exception: + return "" + # A low-confidence fragment ("S", "10", "lohn") is a misread, not + # evidence: the whole crop then counts as unreadable. + if any( + float(getattr(line, "confidence", 0.0) or 0.0) + < PIXEL_CENSUS_OCR_MIN_CONFIDENCE + for line in lines + ): + return "" + return identity_mod.squash( + " ".join(str(getattr(line, "text", "") or "") for line in lines) + ) + + def texts_differ(a: str, b: str) -> bool: + """Clearly different labels, not one label read two ways.""" + import difflib + + if a in b or b in a: + return False # a dropped or clipped word, not another label + ratio = difflib.SequenceMatcher(None, a, b, autojunk=False).ratio() + return ratio < PIXEL_CENSUS_OCR_MAX_SIMILARITY + + target_text: Optional[str] = None + count = 1 + for x, y in peaks: + if is_target(x, y): + continue + crop = identity_mod.crop_region(before_png, (x, y, rw, rh)) + if crop is None: + return census_mod.CensusDecision("uncounted") + verdict = identity_mod.verify_pixel_identity( + recorded, crop, enable_verify=self.pixel_verify_enabled + ) + if verdict is not None and verdict.status == "mismatch": + continue + # The pixels do not rule this place out. Sibling controls that + # share their chrome (section tabs, toggle buttons) look alike; + # their own text still tells them apart. Only when OCR reads both + # crops and the text differs is the place distinguishable. + if target_text is None: + target_text = crop_text(live) + other_text = crop_text(crop) + if target_text and other_text and texts_differ(target_text, other_text): + continue + count += 1 + if count > 1: + return census_mod.CensusDecision("ambiguous", count) + return census_mod.CensusDecision("unique", count) + def _leads_into_consequential( self, step: Step, workflow: Optional[Workflow] ) -> bool: diff --git a/packages/flow/openadapt_flow/vision/match.py b/packages/flow/openadapt_flow/vision/match.py index 251e769a5..933d12c7d 100644 --- a/packages/flow/openadapt_flow/vision/match.py +++ b/packages/flow/openadapt_flow/vision/match.py @@ -309,6 +309,70 @@ def find_template( ) +# A crop that matches itself shifted by a quarter of its width or height at or +# above this correlation (a blank band, a rule, a repeating pattern) matches +# along that axis anywhere: it cannot tell one place from another. Identifier +# crops of real records measured at most ~0.65. +SELF_SIMILAR_SHIFT_SCORE = 0.9 +# Below this grey-level spread a crop is flat. +FLAT_CROP_STD = 2.0 + + +def crop_is_featureless(template_png: bytes) -> bool: + """Whether a crop has too little structure to tell one place from another. + + True for a flat crop, and for one that matches itself shifted by a quarter + of its width or height (a background band, a ruled line, a repeat), so a + template match finds it at every offset along that axis. + """ + crop = _decode_gray(template_png).astype(np.float64) + if crop.size == 0 or float(crop.std()) < FLAT_CROP_STD: + return True + height, width = crop.shape[:2] + + def shifted_score(a: np.ndarray, b: np.ndarray) -> float: + a = a - a.mean() + b = b - b.mean() + energy_a = float((a * a).sum()) + energy_b = float((b * b).sum()) + if energy_a == 0 and energy_b == 0: + return 1.0 # both windows flat: nothing tells places apart + if energy_a == 0 or energy_b == 0: + # Content confined to one edge quarter (a short label in a wide + # cell): the shifted windows are not alike. + return 0.0 + return float((a * b).sum()) / math.sqrt(energy_a * energy_b) + + dx = max(1, width // 4) + dy = max(1, height // 4) + horizontal = shifted_score(crop[:, dx:], crop[:, :-dx]) if width > dx else 1.0 + vertical = shifted_score(crop[dy:, :], crop[:-dy, :]) if height > dy else 1.0 + return max(horizontal, vertical) >= SELF_SIMILAR_SHIFT_SCORE + + +def template_peaks( + screen_png: bytes, template_png: bytes, *, threshold: float = 0.6 +) -> Optional[list[tuple[int, int]]]: + """Every spatially distinct place ``template`` matches, at its own scale. + + Returns the top-left ``(x, y)`` of each non-max-suppressed peak at or above + ``threshold``, strongest first, for a caller that must know every place a + crop could be (the pixel identity census). None when the frame holds at + least :data:`MAX_PEAKS` peaks, so the list may be incomplete, or when the + template does not fit the frame. + """ + screen = _decode_gray(screen_png) + template = _decode_gray(template_png) + th, tw = template.shape[:2] + if th > screen.shape[0] or tw > screen.shape[1]: + return None + result = imgops.match_template_ccoeff_normed(screen, template) + peaks = _peaks_above(result, threshold, tw, th) + if len(peaks) >= MAX_PEAKS: + return None + return [(x, y) for _score, x, y in peaks] + + def find_structural_template( screen_png: bytes, template_png: bytes, diff --git a/packages/flow/tests/native_apps/Worklist.swift b/packages/flow/tests/native_apps/Worklist.swift new file mode 100644 index 000000000..362a33114 --- /dev/null +++ b/packages/flow/tests/native_apps/Worklist.swift @@ -0,0 +1,33 @@ +import AppKit +import Foundation + +struct Row: Decodable { let text: String; let id: String } +let rows = try! JSONDecoder().decode([Row].self, from: CommandLine.arguments[1].data(using: .utf8)!) + +final class Source: NSObject, NSTableViewDataSource, NSTableViewDelegate { + func numberOfRows(in tableView: NSTableView) -> Int { rows.count } + func tableView(_ tableView: NSTableView, viewFor tableColumn: NSTableColumn?, row: Int) -> NSView? { + let field = NSTextField(labelWithString: rows[row].text) + if !rows[row].id.isEmpty { field.identifier = NSUserInterfaceItemIdentifier(rows[row].id) } + return field + } +} + +let app = NSApplication.shared +app.setActivationPolicy(.regular) +let window = NSWindow(contentRect: NSRect(x: 120, y: 120, width: 480, height: 320), + styleMask: [.titled, .closable, .resizable], backing: .buffered, defer: false) +window.title = CommandLine.arguments[2] +let table = NSTableView(frame: NSRect(x: 0, y: 0, width: 480, height: 320)) +let column = NSTableColumn(identifier: NSUserInterfaceItemIdentifier("record")) +column.width = 460 +table.addTableColumn(column) +let source = Source() +table.dataSource = source +table.delegate = source +let scroll = NSScrollView(frame: NSRect(x: 0, y: 0, width: 480, height: 320)) +scroll.documentView = table +window.contentView = scroll +window.makeKeyAndOrderFront(nil) +app.activate(ignoringOtherApps: true) +app.run() diff --git a/packages/flow/tests/test_census_teach_role.py b/packages/flow/tests/test_census_teach_role.py index 9ef238b41..8039043be 100644 --- a/packages/flow/tests/test_census_teach_role.py +++ b/packages/flow/tests/test_census_teach_role.py @@ -22,10 +22,12 @@ from openadapt_flow.runtime.replayer import Replayer from openadapt_flow.vision import Match from tests.test_identity_evidence_capability import ( + _H, _W, _button, _crop_only_step, _gate, + _id_crop, ) from tests.test_replayer import ( FakeBackend, @@ -59,6 +61,21 @@ def _with_role(step, role: str | None): return step +@pytest.fixture() +def no_pixel_census(monkeypatch): + """A surface where no census can count: the pixel census is unavailable too + (for example, the frame cannot be searched for the identifier crop).""" + + from openadapt_flow.runtime import identity_census as census_mod + + monkeypatch.setattr( + Replayer, + "_pixel_census", + lambda self, *a, **k: census_mod.CensusDecision("cannot_count"), + ) + + +@pytest.mark.usefixtures("no_pixel_census") @pytest.mark.parametrize("role", ["button", "textbox", "combobox", "checkbox"]) def test_a_pixel_resolved_form_field_before_a_save_verifies( tmp_path: Path, role: str @@ -74,6 +91,7 @@ def test_a_pixel_resolved_form_field_before_a_save_verifies( assert check.census_status == "not_a_row" +@pytest.mark.usefixtures("no_pixel_census") @pytest.mark.parametrize("role", [None, "row", "listitem", "link", "gridcell"]) def test_a_pixel_resolved_row_that_cannot_be_counted_still_halts( tmp_path: Path, role: str | None @@ -86,6 +104,7 @@ def test_a_pixel_resolved_row_that_cannot_be_counted_still_halts( assert check is not None and check.ambiguity_halt == "identity_uncounted" +@pytest.mark.usefixtures("no_pixel_census") def test_a_row_recorded_at_teach_time_is_counted_even_with_a_control_role( tmp_path: Path, ) -> None: @@ -200,3 +219,51 @@ def test_a_recorded_structured_identity_with_no_teach_census_is_not_a_control() None, structured_identity="John Smith Follow-up Dr Lee" ) assert teach_time_role(step) == "unknown" + + +@pytest.mark.parametrize("role", ["button", None]) +def test_the_pixel_census_decides_for_a_control_when_it_can_count( + tmp_path: Path, role: str | None +) -> None: + """A census that can count always decides, a control role included: the + crop found once verifies, the crop found twice halts.""" + + step, res = _crop_only_step(tmp_path, _id_crop()) + _with_role(step, role) + error, check = _gate(tmp_path, step, res, _id_crop(), pixel_verify_enabled=True) + assert error is None + assert check is not None and check.candidate_count == 1 + + twice = np.vstack([_id_crop(), np.full((8, _W), 255, np.uint8), _id_crop()]) + error, check = _gate(tmp_path, step, res, twice, pixel_verify_enabled=True) + assert error is not None + assert check is not None and check.ambiguity_halt == "identical_candidates" + + +def _blank_band() -> np.ndarray: + """OpenEMR step_016's recorded crop: a dark background band with one + light rule across it, and no label.""" + img = np.full((_H, _W), 40, np.uint8) + img[_H // 2 : _H // 2 + 2, :] = 220 + return img + + +def test_a_blank_band_crop_on_a_control_before_a_save_verifies(tmp_path: Path) -> None: + """The OpenEMR case on #1518: the crop matches all along its band, so the + pixel census cannot count; the recorded button role then applies.""" + + step, res = _crop_only_step(tmp_path, _blank_band()) + _with_role(step, "button") + frame = np.hstack([_blank_band()] * 3) + error, check = _gate(tmp_path, step, res, frame, pixel_verify_enabled=True) + assert error is None + assert check is not None and check.census_status == "not_a_row" + + +def test_a_blank_band_crop_with_no_role_still_halts(tmp_path: Path) -> None: + step, res = _crop_only_step(tmp_path, _blank_band()) + _with_role(step, None) + frame = np.hstack([_blank_band()] * 3) + error, check = _gate(tmp_path, step, res, frame, pixel_verify_enabled=True) + assert error is not None + assert check is not None and check.ambiguity_halt == "identity_uncounted" diff --git a/packages/flow/tests/test_identity_evidence_capability.py b/packages/flow/tests/test_identity_evidence_capability.py index 75d01b6f6..35c9c927c 100644 --- a/packages/flow/tests/test_identity_evidence_capability.py +++ b/packages/flow/tests/test_identity_evidence_capability.py @@ -40,6 +40,16 @@ def _button(val: int = 255) -> np.ndarray: return img +def _id_crop() -> np.ndarray: + """An identifier crop with real structure (text), unlike ``_button``'s + full-height bar, which matches itself shifted and so is featureless.""" + img = np.full((_H, _W), 255, np.uint8) + cv2.putText( + img, "MG4408 Okafor", (6, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.7, 0, 2, cv2.LINE_AA + ) + return img + + def _with_extra_bar(img: np.ndarray) -> np.ndarray: other = img.copy() other[:, 140:145] = 0 @@ -116,21 +126,33 @@ def test_no_surrounding_text_valid_crop_verifies_where_crops_are_verifiable( assert error is None assert check is not None assert check.status == "verified" and check.mode == "pixel" - # A crop alone gives no row text to count look-alikes with; nothing - # consequential follows this step, so it runs and records that. + # A full-height bar matches itself shifted, so it cannot count rows; with + # nothing consequential after the step it runs and records that. assert check.census_status == "cannot_count" -def test_a_verified_crop_with_no_row_count_halts_before_a_write( +def test_a_crop_found_once_on_the_frame_runs_before_a_write( tmp_path: Path, ) -> None: - """The crop matches, but nothing can say no other row shows the same.""" + """No OCR band: the census counts where the crop verifies. Once: it runs.""" - step, res = _crop_only_step(tmp_path, _button()) - error, check = _gate(tmp_path, step, res, _button(), pixel_verify_enabled=True) - assert error is not None and "identity_uncounted" in error + step, res = _crop_only_step(tmp_path, _id_crop()) + error, check = _gate(tmp_path, step, res, _id_crop(), pixel_verify_enabled=True) + assert error is None + assert check is not None and check.status == "verified" + assert check.candidate_count == 1 + + +def test_a_crop_found_twice_on_the_frame_halts(tmp_path: Path) -> None: + """Another row shows the same identifier pixels: the step stops.""" + + step, res = _crop_only_step(tmp_path, _id_crop()) + frame = np.vstack([_id_crop(), np.full((8, _W), 255, np.uint8), _id_crop()]) + error, check = _gate(tmp_path, step, res, frame, pixel_verify_enabled=True) + assert error is not None and "2 shown rows" in error assert check is not None - assert check.status == "abstain" and check.ambiguity_halt == "identity_uncounted" + assert check.status == "abstain" and check.ambiguity_halt == "identical_candidates" + assert check.candidate_count == 2 def test_crop_that_does_not_match_halts(tmp_path: Path) -> None: diff --git a/packages/flow/tests/test_native_census_real_apps.py b/packages/flow/tests/test_native_census_real_apps.py new file mode 100644 index 000000000..e14b76900 --- /dev/null +++ b/packages/flow/tests/test_native_census_real_apps.py @@ -0,0 +1,331 @@ +"""Row census against real native list apps (opt-in, one per platform). + +Each test starts a small native app showing a worklist, reads it through the +backend's real accessibility code, and checks the census: identical rows are +ambiguous, distinct record ids pick the recorded row, a unique row runs. + +* Windows: a WPF ``ListBox`` whose items carry ``AutomationProperties + .AutomationId``, read through the guest agent's UI Automation census. + Set ``OPENADAPT_TEST_WINDOWS_UIA=1``; needs the ``uiautomation`` package. +* macOS: a view-based ``NSTableView`` (tests/native_apps/Worklist.swift, built + with ``swiftc``) whose cells carry an identifier, read through + ``QuartzMacAXClient.row_census``. Set ``OPENADAPT_TEST_MACOS_AX=1``; needs + PyObjC and Accessibility access for the test process. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from pathlib import Path +from typing import Any, Optional + +import pytest + +from openadapt_flow.runtime import identity_template as itmpl +from openadapt_flow.runtime.identity_census import CensusDecision, decide + +pytestmark = pytest.mark.timeout(180) + +TITLE = "OA Census Worklist" +SAME = "John Smith Follow-up Dr Lee" +OTHER = "Maria Garcia Intake Dr Lee" + + +def _decide(census: Optional[dict], recorded_key: Optional[str] = None): + tmpl = itmpl.build_identity_template( + None, structured_identity="x", record_key=recorded_key + ) + assert tmpl is not None + return decide( + census, + has_recorded_key=tmpl.record_key is not None, + key_matches=lambda key: itmpl.record_key_matches(tmpl, key), + ) + + +def _wait(predicate, timeout_s: float = 30.0) -> Any: + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + value = predicate() + if value: + return value + time.sleep(0.5) + raise AssertionError("the native test app did not appear") + + +# -- Windows (WPF + UI Automation) --------------------------------------------- + +_WPF_APP = r""" +param([string]$RowsJson) +Add-Type -AssemblyName PresentationFramework +$rows = $RowsJson | ConvertFrom-Json +$window = New-Object System.Windows.Window +$window.Title = '__TITLE__' +$window.Width = 520 +$window.Height = 360 +$window.Left = 40 +$window.Top = 40 +$list = New-Object System.Windows.Controls.ListBox +foreach ($row in $rows) { + $item = New-Object System.Windows.Controls.ListBoxItem + $panel = New-Object System.Windows.Controls.StackPanel + $panel.Orientation = 'Horizontal' + foreach ($cell in $row.cells) { + $text = New-Object System.Windows.Controls.TextBlock + $text.Text = $cell + $text.Margin = '0,0,24,0' + [void]$panel.Children.Add($text) + } + $item.Content = $panel + if ($row.id) { + [System.Windows.Automation.AutomationProperties]::SetAutomationId($item, $row.id) + } + [void]$list.Items.Add($item) +} +$window.Content = $list +[void]$window.ShowDialog() +""".replace("__TITLE__", TITLE) + + +@pytest.fixture +def wpf_worklist(tmp_path: Path): + if sys.platform != "win32" or os.environ.get("OPENADAPT_TEST_WINDOWS_UIA") != "1": + pytest.skip("set OPENADAPT_TEST_WINDOWS_UIA=1 on Windows") + auto = pytest.importorskip("uiautomation") + script = tmp_path / "worklist.ps1" + script.write_text(_WPF_APP, encoding="utf-8") + procs: list[subprocess.Popen] = [] + + def start(rows: list[dict[str, Any]]) -> list[Any]: + proc = subprocess.Popen( + [ + "powershell", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + str(script), + "-RowsJson", + json.dumps(rows), + ] + ) + procs.append(proc) + window = _wait( + lambda: ( + auto.WindowControl(searchDepth=1, Name=TITLE) + if auto.WindowControl(searchDepth=1, Name=TITLE).Exists(0, 0) + else None + ) + ) + listing = window.ListControl(searchDepth=3) + return _wait( + lambda: ( + [ + c + for c in listing.GetChildren() + if c.ControlTypeName == "ListItemControl" + ] + if listing.Exists(0, 0) + else None + ) + ) + + yield auto, start + for proc in procs: + proc.kill() + + +def _first_cell_center(item: Any) -> tuple[int, int]: + cell = item.GetChildren()[0] + rect = cell.BoundingRectangle + return (rect.left + rect.right) // 2, (rect.top + rect.bottom) // 2 + + +def test_windows_wpf_identical_rows_are_ambiguous(wpf_worklist) -> None: + from openadapt_flow.backends.win_agent import server + + auto, start = wpf_worklist + items = start( + [ + {"cells": SAME.split(" ", 2), "id": ""}, + {"cells": SAME.split(" ", 2), "id": ""}, + {"cells": OTHER.split(" ", 2), "id": ""}, + ] + ) + with auto.UIAutomationInitializerInThread(): + census = server._identity_candidates_at(auto, *_first_cell_center(items[1])) + assert _decide(census) == CensusDecision("ambiguous", 2), census + + +def test_windows_wpf_record_ids_pick_the_recorded_row(wpf_worklist) -> None: + from openadapt_flow.backends.win_agent import server + + auto, start = wpf_worklist + items = start( + [ + {"cells": SAME.split(" ", 2), "id": "chart-1002"}, + {"cells": SAME.split(" ", 2), "id": "chart-1001"}, + ] + ) + with auto.UIAutomationInitializerInThread(): + census = server._identity_candidates_at(auto, *_first_cell_center(items[0])) + unique = server._identity_candidates_at(auto, *_first_cell_center(items[1])) + decision = _decide(census, "automation-id=chart-1001") + assert decision.verdict == "retarget", census + assert decision.offset is not None and decision.offset[1] > 0 + assert _decide(unique, "automation-id=chart-1001").verdict == "unique" + + +def test_windows_wpf_unique_row_runs(wpf_worklist) -> None: + from openadapt_flow.backends.win_agent import server + + auto, start = wpf_worklist + items = start( + [ + {"cells": SAME.split(" ", 2), "id": ""}, + {"cells": OTHER.split(" ", 2), "id": ""}, + ] + ) + with auto.UIAutomationInitializerInThread(): + point = _first_cell_center(items[1]) + census = server._identity_candidates_at(auto, *point) + text = server._structured_text_at(auto, *point) + assert text, "the identity check reads no row text from this list" + assert _decide(census) == CensusDecision("unique", 1), census + + +# -- macOS (NSTableView + AX) -------------------------------------------------- + +_SWIFT_APP = Path(__file__).parent / "native_apps" / "Worklist.swift" + + +def _on_screen(pid: int) -> bool: + import Quartz + + windows = Quartz.CGWindowListCopyWindowInfo( + Quartz.kCGWindowListOptionOnScreenOnly, Quartz.kCGNullWindowID + ) + return any( + w.get("kCGWindowOwnerPID") == pid and w.get("kCGWindowName") == TITLE + for w in windows + ) + + +@pytest.fixture +def cocoa_worklist(tmp_path: Path): + if sys.platform != "darwin" or os.environ.get("OPENADAPT_TEST_MACOS_AX") != "1": + pytest.skip("set OPENADAPT_TEST_MACOS_AX=1 on macOS") + services = pytest.importorskip("ApplicationServices") + pytest.importorskip("Quartz") + if not services.AXIsProcessTrusted(): + pytest.skip("this process has no Accessibility access") + from openadapt_flow.backends.macos_backend import QuartzMacAXClient, _AXTreeView + + binary = tmp_path / "worklist" + built = subprocess.run( + ["swiftc", "-O", "-o", str(binary), str(_SWIFT_APP)], + capture_output=True, + text=True, + timeout=170, + ) + if built.returncode != 0: + pytest.skip(f"swiftc could not build the table app: {built.stderr[-400:]}") + procs: list[subprocess.Popen] = [] + client = QuartzMacAXClient() + + def start(rows: list[dict[str, str]]) -> tuple[int, list[tuple[float, float]]]: + proc = subprocess.Popen([str(binary), json.dumps(rows), TITLE]) + procs.append(proc) + view = _AXTreeView(client) + + def row_centers() -> Optional[list[tuple[float, float]]]: + window = client._window_element(proc.pid, TITLE) + if window is None: + return None + stack, found = [window], [] + while stack: + node = stack.pop() + if view.role(node) == "AXRow": + bounds = client._bounds(node) + if bounds is not None: + found.append((bounds[0] + 20, bounds[1] + bounds[3] / 2)) + continue + stack.extend(view.children(node)) + return ( + sorted(found, key=lambda p: p[1]) if len(found) == len(rows) else None + ) + + try: + return proc.pid, _wait(row_centers) + except AssertionError: + if _on_screen(proc.pid): + # The window is on screen but AX lists no window for it: this + # session's Accessibility access is degraded (AX answers with + # the application element), so nothing can be read. + pytest.skip("AX lists no windows in this session; cannot read the app") + raise + + yield client, start + for proc in procs: + proc.kill() + + +def _ax_trail(client: Any, pid: int, x: float, y: float) -> list[Any]: + """Roles and text from the element at (x, y) up to its window, for a + failure message.""" + + from ApplicationServices import ( + AXUIElementCopyElementAtPosition, + AXUIElementCreateApplication, + ) + + from openadapt_flow.backends.macos_backend import _AXTreeView + + view = _AXTreeView(client) + _err, node = AXUIElementCopyElementAtPosition( + AXUIElementCreateApplication(pid), float(x), float(y), None + ) + trail = [] + for _ in range(10): + if node is None: + break + trail.append((view.role(node), client._text(node), view.id_attributes(node))) + node = view.parent(node) + return trail + + +def test_macos_table_identical_rows_are_ambiguous(cocoa_worklist) -> None: + client, start = cocoa_worklist + pid, centers = start( + [{"text": SAME, "id": ""}, {"text": SAME, "id": ""}, {"text": OTHER, "id": ""}] + ) + census = client.row_census(pid, TITLE, *centers[1], limit=4000) + assert _decide(census) == CensusDecision("ambiguous", 2), ( + census, + _ax_trail(client, pid, *centers[1]), + ) + + +def test_macos_table_record_ids_pick_the_recorded_row(cocoa_worklist) -> None: + client, start = cocoa_worklist + pid, centers = start( + [{"text": SAME, "id": "chart-1002"}, {"text": SAME, "id": "chart-1001"}] + ) + census = client.row_census(pid, TITLE, *centers[0], limit=4000) + decision = _decide(census, "ax-identifier=chart-1001") + assert decision.verdict == "retarget", (census, _ax_trail(client, pid, *centers[0])) + assert decision.offset is not None and decision.offset[1] > 0 + + +def test_macos_table_unique_row_runs(cocoa_worklist) -> None: + client, start = cocoa_worklist + pid, centers = start([{"text": SAME, "id": ""}, {"text": OTHER, "id": ""}]) + census = client.row_census(pid, TITLE, *centers[1], limit=4000) + assert _decide(census) == CensusDecision("unique", 1), ( + census, + _ax_trail(client, pid, *centers[1]), + ) diff --git a/packages/flow/tests/test_native_identity_census.py b/packages/flow/tests/test_native_identity_census.py new file mode 100644 index 000000000..7f014b862 --- /dev/null +++ b/packages/flow/tests/test_native_identity_census.py @@ -0,0 +1,1001 @@ +"""Row census on native accessibility trees (macOS AX, Linux AT-SPI, Windows UIA). + +A worklist of two rows that read "John Smith Follow-up Dr Lee" must halt +before it opens either chart, distinct record ids must pick the recorded row, +and a unique row must still run. Each backend is exercised through its real +census code over a fake accessibility tree; the browser census has its own +real-browser tests (tests/test_realism_browser.py). +""" + +from __future__ import annotations + +import types +import uuid +from typing import Any, Optional + +import pytest + +from openadapt_flow.backends import linux_backend +from openadapt_flow.backends.win_agent import server +from openadapt_flow.ir import ( + ActionKind, + Anchor, + IdentityCheck, + Resolution, + Step, + Workflow, +) +from openadapt_flow.runtime import identity_template as itmpl +from openadapt_flow.runtime.identity_census import ( + CensusDecision, + decide, + tree_row_census, +) +from openadapt_flow.runtime.replayer import Replayer + +ROW = ("John Smith", "Follow-up", "Dr Lee") +OTHER = ("Maria Garcia", "Intake", "Dr Lee") + + +def _decide(census: Optional[dict], recorded_key: Optional[str] = None): + tmpl = itmpl.build_identity_template( + None, structured_identity="x", record_key=recorded_key + ) + assert tmpl is not None + return decide( + census, + has_recorded_key=tmpl.record_key is not None, + key_matches=lambda key: itmpl.record_key_matches(tmpl, key), + ) + + +# -- generic tree census ------------------------------------------------------- + + +class Node: + def __init__(self, role: str, text: str = "", *, ids=(), y: float = 0.0): + self.role = role + self.text = text + self.ids = list(ids) + self.y = y + self.parent: Optional[Node] = None + self.kids: list[Node] = [] + self.declared_rows: Optional[int] = None + self.scrolls: Optional[bool] = False + + def add(self, *children: "Node") -> "Node": + for child in children: + child.parent = self + self.kids.append(child) + return self + + +class View: + def parent(self, node: Node) -> Any: + return node.parent + + def children(self, node: Node) -> list[Any]: + return list(node.kids) + + def role(self, node: Node) -> Optional[str]: + return node.role + + def origin(self, node: Node) -> Optional[tuple[float, float]]: + return (0.0, node.y) + + def id_attributes(self, node: Node) -> list[tuple[str, str]]: + return [("id", value) for value in node.ids] + + def row_count(self, container: Node) -> Optional[int]: + return container.declared_rows + + def scrollable(self, container: Node) -> Optional[bool]: + return container.scrolls + + +def _table(*rows: tuple[tuple[str, ...], tuple[str, ...]]) -> Node: + """A table of rows ``(cells, row ids)``, 40 px apart.""" + + table = Node("table") + for index, (cells, ids) in enumerate(rows): + row = Node("row", ids=ids, y=40.0 * index) + row.add(*(Node("cell", text) for text in cells)) + table.add(row) + return table + + +def _row_identity(row: Node, path: list[int]) -> Optional[str]: + """Row text except the target's cell (the Windows/Linux reading).""" + + return " ".join(c.text for i, c in enumerate(row.kids) if i != path[0]) or None + + +def _census(target: Node) -> Optional[dict]: + return tree_row_census( + target, View(), row_roles=frozenset({"row"}), identity=_row_identity + ) + + +def test_two_identical_rows_are_ambiguous() -> None: + table = _table((ROW, ()), (ROW, ()), (OTHER, ())) + census = _census(table.kids[1].kids[0]) + assert census is not None and len(census["candidates"]) == 2 + assert _decide(census) == CensusDecision("ambiguous", 2) + + +def test_three_identical_rows_are_ambiguous() -> None: + table = _table((ROW, ()), (OTHER, ()), (ROW, ()), (ROW, ())) + assert _decide(_census(table.kids[0].kids[0])) == CensusDecision("ambiguous", 3) + + +def test_distinct_record_ids_pick_the_recorded_row() -> None: + table = _table((ROW, ("chart-1002",)), (ROW, ("chart-1001",))) + census = _census(table.kids[0].kids[0]) # resolver landed on 1002 + assert _decide(census, "id=chart-1001") == CensusDecision( + "retarget", 2, (0.0, 40.0) + ) + assert _decide(_census(table.kids[1].kids[0]), "id=chart-1001") == ( + CensusDecision("unique", 2) + ) + + +def test_a_unique_row_runs() -> None: + table = _table((ROW, ()), (OTHER, ())) + assert _decide(_census(table.kids[1].kids[2])) == CensusDecision("unique", 1) + + +def test_a_control_outside_any_row_is_not_a_row() -> None: + window = Node("window").add(Node("button", "Sign visit note")) + assert _census(window.kids[0]) == {"row": False} + + +@pytest.mark.parametrize( + "ids", + [ + ("row-0", "row-1"), + ("r1c2", "r2c2"), + ("row-3-col-1", "row-4-col-1"), + ("row-53", "row-54"), # absolute positions in a longer list + ("item_12", "item_13"), + ("0", "1"), + ], +) +def test_positional_ids_never_tell_rows_apart(ids) -> None: + table = _table((ROW, (ids[0],)), (ROW, (ids[1],))) + census = _census(table.kids[0].kids[0]) + assert census is not None + assert [c["record_key"] for c in census["candidates"]] == [None, None] + assert _decide(census).verdict == "ambiguous" + + +_STABLE_ID_FAMILIES = { + "uuid4": lambda r: str(uuid.UUID(int=r.getrandbits(128), version=4)), + "uuid4 without dashes": lambda r: uuid.UUID(int=r.getrandbits(128)).hex, + "encounter uuid": lambda r: f"encounter-{uuid.UUID(int=r.getrandbits(128))}", + "MRN-A1234": lambda r: f"MRN-A{r.randint(1000, 99999)}", + "zero-padded MRN": lambda r: f"{r.randint(1, 99_999_999):08d}", + "chart-1001": lambda r: f"chart-{r.randint(1000, 999_999)}", + "patient_N": lambda r: f"patient_{r.randint(100, 999_999)}", + "ENC2024-000123": lambda r: ( + f"ENC{r.randint(2015, 2026)}-{r.randint(1, 999_999):06d}" + ), + "FHIR Patient/123": lambda r: f"Patient/{r.randint(1, 99_999)}", + "dated visit": lambda r: f"visit-2026-10-{r.randint(10, 28)}-{r.randint(1, 99)}", + "ObjectId": lambda r: "%024x" % r.getrandbits(96), + "ULID": lambda r: ( + "01H" + "".join(r.choice("0123456789ABCDEFGHJKMNPQRSTVWXYZ") for _ in range(23)) + ), +} + + +@pytest.mark.parametrize("family", sorted(_STABLE_ID_FAMILIES)) +def test_stable_record_ids_are_never_dropped_as_positional(family) -> None: + """Measured: 0 of 1000 ids per family read as a position.""" + + import random + + from openadapt_flow.runtime.identity_census import positional_id + + rng = random.Random(family) + values = [_STABLE_ID_FAMILIES[family](rng) for _ in range(1000)] + dropped = [v for i, v in enumerate(values) if positional_id(v, i % 50)] + assert dropped == [] + + +def test_the_windows_agent_drops_the_same_ids() -> None: + import random + + from openadapt_flow.runtime.identity_census import positional_id + + rng = random.Random(3) + values = [g(rng) for g in _STABLE_ID_FAMILIES.values() for _ in range(50)] + values += ["row-3", "r3c2", "item_12", "row-3-col-1", "3-1", "ListViewItem-0", "7"] + for index, value in enumerate(values): + assert server._positional_id(value, index % 9) == positional_id( + value, index % 9 + ), value + + +@pytest.mark.parametrize("scrolls", [True, None]) +def test_a_list_without_a_row_count_that_may_scroll_is_incomplete(scrolls) -> None: + table = _table((ROW, ()), (OTHER, ())) + table.scrolls = scrolls # no row count, and rows may be out of view + census = _census(table.kids[1].kids[0]) + assert census is not None and census["incomplete"] is True + + +def test_a_table_with_more_rows_than_it_exposes_is_incomplete() -> None: + table = _table((ROW, ()), (OTHER, ())) + table.declared_rows = 50 # a virtualized list: rows built on scroll + census = _census(table.kids[1].kids[0]) + assert census is not None and census["incomplete"] is True + assert _decide(census).verdict == "uncounted" + + +def test_a_row_whose_cells_are_not_built_is_incomplete() -> None: + table = _table((ROW, ()), (OTHER, ())) + table.add(Node("row", y=80.0)) # placeholder row, no cells yet + census = _census(table.kids[0].kids[2]) + assert census is not None and census["incomplete"] is True + + +def test_list_items_without_a_row_role_are_rows_of_their_list() -> None: + """macOS AXList: the list's direct children are its rows.""" + + listing = Node("AXList") + for index, name in enumerate(("John Smith", "John Smith", "Maria Garcia")): + item = Node("AXGroup", y=30.0 * index).add(Node("AXStaticText", name)) + listing.add(item) + + def own_text(row: Node, path: list[int]) -> Optional[str]: + node = row + for step in path: + node = node.kids[step] + return node.text or None + + census = tree_row_census( + listing.kids[2].kids[0], + View(), + row_roles=frozenset({"AXRow"}), + container_roles=frozenset({"AXList"}), + identity=own_text, + ) + assert _decide(census) == CensusDecision("unique", 1) + census = tree_row_census( + listing.kids[0].kids[0], + View(), + row_roles=frozenset({"AXRow"}), + container_roles=frozenset({"AXList"}), + identity=own_text, + ) + assert _decide(census) == CensusDecision("ambiguous", 2) + + +# -- Linux AT-SPI client ------------------------------------------------------- + + +class _States: + def __init__(self, names: set[str]): + self._names = names + + def contains(self, state: str) -> bool: + return state in self._names + + +class _Extents: + def __init__(self, x: int, y: int, w: int, h: int): + self.x, self.y, self.width, self.height = x, y, w, h + + +class AtspiNode: + def __init__(self, role: str, name: str = "", *, bounds=(0, 0, 10, 10), ident=""): + self._role = role + self._name = name + self._bounds = bounds + self._ident = ident + self._parent: Optional[AtspiNode] = None + self._kids: list[AtspiNode] = [] + + def add(self, *kids: "AtspiNode") -> "AtspiNode": + for kid in kids: + kid._parent = self + self._kids.append(kid) + return self + + def get_role_name(self) -> str: + return self._role + + def get_name(self) -> str: + return self._name + + def get_parent(self): + return self._parent + + def get_child_count(self) -> int: + return len(self._kids) + + def get_child_at_index(self, index: int): + return self._kids[index] + + def get_state_set(self) -> _States: + return _States({"VISIBLE", "SHOWING", "ENABLED"}) + + def get_component_iface(self): + node = self + + class _Component: + def get_extents(self, _coord): + return _Extents(*node._bounds) + + return _Component() + + def get_accessible_id(self) -> str: + return self._ident + + +def _atspi_client() -> linux_backend.AtspiLinuxClient: + client = object.__new__(linux_backend.AtspiLinuxClient) + client._atspi = types.SimpleNamespace( + StateType=types.SimpleNamespace( + VISIBLE="VISIBLE", SHOWING="SHOWING", ENABLED="ENABLED", FOCUSED="FOCUSED" + ), + CoordType=types.SimpleNamespace(SCREEN="SCREEN"), + ) + return client + + +def _atspi_window(*rows: tuple[tuple[str, ...], str]): + listing = AtspiNode("list", bounds=(0, 0, 400, 300)) + for index, (cells, ident) in enumerate(rows): + top = 40 * index + row = AtspiNode("list item", bounds=(0, top, 400, 40), ident=ident) + for column, text in enumerate(cells): + row.add(AtspiNode("label", text, bounds=(100 * column, top, 100, 40))) + listing.add(row) + frame = AtspiNode("frame", "Worklist", bounds=(0, 0, 400, 300)).add(listing) + return linux_backend.LinuxWindow( + "w1", "app", "Worklist", 1, (0, 0, 400, 300), frame + ) + + +def test_linux_identical_rows_are_ambiguous() -> None: + window = _atspi_window((ROW, ""), (ROW, ""), (OTHER, "")) + census = _atspi_client().row_census(window, 150, 50, limit=100) + assert _decide(census) == CensusDecision("ambiguous", 2) + + +def test_linux_record_ids_pick_the_recorded_row() -> None: + window = _atspi_window((ROW, "chart-1002"), (ROW, "chart-1001")) + census = _atspi_client().row_census(window, 150, 10, limit=100) + assert _decide(census, "atspi-id=chart-1001") == CensusDecision( + "retarget", 2, (0.0, 40.0) + ) + + +def test_linux_unique_row_runs_and_matches_structured_text() -> None: + client = _atspi_client() + window = _atspi_window((ROW, ""), (OTHER, "")) + element = client.element_at_point(window, 150, 50) + assert element is not None + # The census reads rows exactly as the identity check does. + assert client.structured_text(element) == "Maria Garcia Dr Lee" + census = client.row_census(window, 150, 50, limit=100) + assert _decide(census) == CensusDecision("unique", 1) + + +# -- Windows UIA agent --------------------------------------------------------- + + +class _Rect: + def __init__(self, left: int, top: int, right: int, bottom: int): + self.left, self.top, self.right, self.bottom = left, top, right, bottom + + +class UiaControl: + def __init__(self, control_type: str, name: str = "", *, aid: str = "", top=0): + self.ControlTypeName = control_type + self.Name = name + self.AutomationId = aid + self.BoundingRectangle = _Rect(0, top, 400, top + 40) + self.parent: Optional[UiaControl] = None + self.children: list[UiaControl] = [] + self.grid_rows: Optional[int] = None + self.virtual = False + self.broken = False + + def add(self, *kids: "UiaControl") -> "UiaControl": + for kid in kids: + kid.parent = self + self.children.append(kid) + return self + + def GetParentControl(self): + return self.parent + + def GetChildren(self): + return list(self.children) + + def GetGridPattern(self): + if self.broken: + raise RuntimeError("provider failed") + if self.grid_rows is None: + return None # unsupported, as uiautomation reports it + return types.SimpleNamespace(RowCount=self.grid_rows) + + def GetScrollPattern(self): + return types.SimpleNamespace(VerticallyScrollable=self.virtual) + + def GetItemContainerPattern(self): + return object() if self.virtual else None + + +def _uia_list(*rows: tuple[tuple[str, ...], str]): + listing = UiaControl("ListControl") + for index, (cells, aid) in enumerate(rows): + item = UiaControl("DataItemControl", aid=aid, top=40 * index) + item.add(*(UiaControl("TextControl", text, top=40 * index) for text in cells)) + listing.add(item) + return listing + + +def _auto(target: UiaControl): + return types.SimpleNamespace(ControlFromPoint=lambda _x, _y: target) + + +def test_windows_identical_rows_are_ambiguous() -> None: + listing = _uia_list((ROW, ""), (ROW, ""), (ROW, "")) + census = server._identity_candidates_at( + _auto(listing.children[1].children[0]), 5, 45 + ) + assert _decide(census) == CensusDecision("ambiguous", 3) + + +def test_windows_automation_ids_pick_the_recorded_row() -> None: + listing = _uia_list((ROW, "chart-1002"), (ROW, "chart-1001")) + census = server._identity_candidates_at( + _auto(listing.children[0].children[0]), 5, 5 + ) + assert _decide(census, "automation-id=chart-1001") == CensusDecision( + "retarget", 2, (0.0, 40.0) + ) + + +def test_windows_positional_automation_ids_do_not_tell_rows_apart() -> None: + listing = _uia_list((ROW, "ListViewItem-0"), (ROW, "ListViewItem-1")) + census = server._identity_candidates_at( + _auto(listing.children[0].children[0]), 5, 5 + ) + assert _decide(census).verdict == "ambiguous" + + +def test_windows_unique_row_runs_and_matches_structured_text() -> None: + listing = _uia_list((ROW, ""), (OTHER, "")) + target = listing.children[1].children[0] + auto = _auto(target) + census = server._identity_candidates_at(auto, 5, 45) + assert _decide(census) == CensusDecision("unique", 1) + # The census reads rows exactly as the identity check does: the clicked + # cell is left out. + assert server._structured_text_at(auto, 5, 45) == "Intake Dr Lee" + + +def test_windows_virtualized_or_short_grids_are_incomplete() -> None: + listing = _uia_list((ROW, ""), (OTHER, "")) + listing.grid_rows = 300 + census = server._identity_candidates_at( + _auto(listing.children[1].children[0]), 5, 45 + ) + assert _decide(census).verdict == "uncounted" + listing.grid_rows = None + listing.virtual = True + census = server._identity_candidates_at( + _auto(listing.children[1].children[0]), 5, 45 + ) + assert _decide(census).verdict == "uncounted" + + +def test_windows_a_failing_provider_is_incomplete() -> None: + listing = _uia_list((ROW, ""), (OTHER, "")) + listing.broken = True + census = server._identity_candidates_at( + _auto(listing.children[1].children[0]), 5, 45 + ) + assert _decide(census).verdict == "uncounted" + + +def test_windows_a_row_without_cells_is_incomplete() -> None: + listing = _uia_list((ROW, ""), (OTHER, "")) + listing.add(UiaControl("DataItemControl", top=80)) # not built yet + census = server._identity_candidates_at( + _auto(listing.children[0].children[0]), 5, 5 + ) + assert _decide(census).verdict == "uncounted" + + +def test_windows_standalone_control_is_not_a_row() -> None: + window = UiaControl("WindowControl").add(UiaControl("ButtonControl", "Sign")) + census = server._identity_candidates_at(_auto(window.children[0]), 5, 5) + assert _decide(census).verdict == "not_a_row" + + +def test_windows_agent_serves_the_census_operation() -> None: + listing = _uia_list((ROW, ""), (ROW, "")) + reply = server._perform_uia_initialized( + _auto(listing.children[0].children[0]), + "identity-candidates-at", + {"x": 5, "y": 5}, + ) + assert reply["status"] == "ok" and len(reply["census"]["candidates"]) == 2 + + +# -- macOS backend ------------------------------------------------------------- + + +def test_macos_backend_converts_row_offsets_to_pixels() -> None: + from tests.test_macos_structural import FakeMacAXClient, backend + + class CensusAX(FakeMacAXClient): + def row_census(self, pid, window_title, x, y, *, limit): + return { + "record_key": None, + "incomplete": False, + "top_level": True, + "candidates": [ + {"self": True, "record_key": None, "dx": 0.0, "dy": 0.0}, + {"self": False, "record_key": None, "dx": 0.0, "dy": 22.0}, + ], + } + + target = backend(ax_client=CensusAX()) + target.screenshot() + census = target.identity_candidates_at(10, 10) + assert census is not None + assert census["candidates"][1]["dy"] == pytest.approx(22.0 * (target._scale_y or 1)) + assert _decide(census).verdict == "ambiguous" + + +def test_macos_client_without_a_census_cannot_count() -> None: + from tests.test_macos_structural import FakeMacAXClient, backend + + assert backend(ax_client=FakeMacAXClient()).identity_candidates_at(10, 10) is None + + +# -- the replayer on a native backend ------------------------------------------ + + +class NativeBackend: + """Structured text and a row census over the fake tree, like a native backend.""" + + viewport = (400, 300) + + def __init__(self, table: Node, at: dict[tuple[int, int], Node]): + self._table = table + self._at = at + + def screenshot(self) -> bytes: + return b"" + + def structured_text_at(self, x: int, y: int) -> Optional[str]: + node = self._at.get((x, y)) + if node is None: + return None + row = node.parent + return _row_identity(row, [row.kids.index(node)]) + + def identity_candidates_at(self, x: int, y: int) -> Optional[dict]: + node = self._at.get((x, y)) + return _census(node) if node is not None else None + + +def _step(recorded_key: Optional[str]) -> Step: + template = itmpl.build_identity_template( + None, + structured_identity="Follow-up Dr Lee", + record_key=recorded_key, + structured_row=True, + ) + return Step( + id="open", + intent="click 'John Smith'", + action=ActionKind.CLICK, + identity_armed=True, + anchor=Anchor( + template="templates/open.png", + region=(0, 0, 100, 40), + click_point=(50, 20), + identity_template=template, + ), + ) + + +def _sign() -> Step: + return Step(id="sign", intent="click 'Sign visit note'", action=ActionKind.CLICK) + + +def _verify(backend: NativeBackend, step: Step, point) -> IdentityCheck: + replayer = Replayer(backend, vision=object(), poll_interval_s=0.01) + resolution = Resolution(rung="template", point=point, confidence=1.0, elapsed_ms=1) + return replayer._verify_identity( + step, resolution, b"", {}, Workflow(name="wf", steps=[step, _sign()]) + ) + + +def _native(rows) -> tuple[NativeBackend, Node]: + table = _table(*rows) + at = {(50, 20 + 40 * i): row.kids[0] for i, row in enumerate(table.kids)} + return NativeBackend(table, at), table + + +def test_replay_halts_on_identical_native_rows() -> None: + backend, _ = _native([(ROW, ()), (ROW, ())]) + check = _verify(backend, _step(None), (50, 20)) + assert check.status == "abstain" + assert check.ambiguity_halt == "identical_candidates" + assert check.candidate_count == 2 + + +def test_replay_moves_to_the_native_row_with_the_recorded_id() -> None: + backend, _ = _native([(ROW, ("chart-1002",)), (ROW, ("chart-1001",))]) + step = _step("id=chart-1001") + replayer = Replayer(backend, vision=object(), poll_interval_s=0.01) + resolution = Resolution( + rung="template", point=(50, 20), confidence=1.0, elapsed_ms=1 + ) + moved, _region = replayer._retarget_by_record_key(step, resolution, (0, 0, 100, 40)) + assert moved.point == (50, 60) + check = _verify(backend, step, moved.point) + assert check.status == "verified" and check.candidate_count == 2 + + +def test_replay_runs_a_unique_native_row() -> None: + backend, _ = _native([(ROW, ()), (OTHER, ())]) + check = _verify(backend, _step(None), (50, 20)) + assert check.status == "verified" and check.candidate_count == 1 + + +# -- pixel census (remote display, no OCR band) --------------------------------- + + +def _crop_png(bar: int = 60, height: int = 48, width: int = 240) -> Any: + import numpy as np + + img = np.full((height, width), 255, np.uint8) + img[8:40, bar : bar + 6] = 0 + img[8:40, bar + 40 : bar + 44] = 0 + return img + + +def _png(arr: Any) -> bytes: + import cv2 + + ok, buf = cv2.imencode(".png", arr) + assert ok + return buf.tobytes() + + +def test_template_peaks_lists_every_place_and_refuses_a_crowded_frame() -> None: + import numpy as np + + from openadapt_flow.vision.match import MAX_PEAKS, template_peaks + + crop = _crop_png() + gap = np.full((10, crop.shape[1]), 255, np.uint8) + frame = np.vstack([crop, gap, crop, gap, _crop_png(bar=150)]) + peaks = template_peaks(_png(frame), _png(crop), threshold=0.9) + assert peaks is not None and sorted(y for _x, y in peaks) == [0, 58] + crowded = np.vstack([np.vstack([crop, gap])] * (MAX_PEAKS + 1)) + assert template_peaks(_png(crowded), _png(crop), threshold=0.9) is None + assert template_peaks(_png(crop[:10, :10]), _png(crop)) is None # too small + + +def _pixel_step(tmp_path, recorded: bytes) -> Step: + (tmp_path / "templates").mkdir(parents=True, exist_ok=True) + (tmp_path / "templates" / "id.png").write_bytes(recorded) + (tmp_path / "templates" / "t.png").write_bytes(recorded) + return Step( + id="open", + intent="click 'Open referral'", + action=ActionKind.CLICK, + identity_armed=True, + anchor=Anchor( + template="templates/t.png", + region=(0, 0, 240, 48), + click_point=(0, 0), + identifier_crop="templates/id.png", + identifier_region=(0, 0, 240, 48), + ), + ) + + +class _PixelBackend: + viewport = (240, 400) + + def screenshot(self) -> bytes: + return b"" + + +def _pixel_census( + tmp_path, frame: Any, *, verify: bool, monkeypatch=None, verdict=None +): + from openadapt_flow.runtime import identity as identity_mod + + if verdict is not None: + assert monkeypatch is not None + monkeypatch.setattr( + identity_mod, "verify_pixel_identity", lambda *a, **k: verdict + ) + replayer = Replayer( + _PixelBackend(), + vision=object(), + poll_interval_s=0.01, + pixel_verify_enabled=verify, + ) + step = _pixel_step(tmp_path, _png(_crop_png())) + resolution = Resolution(rung="template", point=(0, 0), confidence=1.0, elapsed_ms=1) + return replayer._pixel_census( + step, resolution, _png(frame), Workflow(name="wf"), tmp_path + ) + + +def _two_copies() -> Any: + import numpy as np + + gap = np.full((10, 240), 255, np.uint8) + return np.vstack([_crop_png(), gap, _crop_png()]) + + +def test_pixel_census_counts_a_second_copy(tmp_path) -> None: + decision = _pixel_census(tmp_path, _two_copies(), verify=True) + assert decision == CensusDecision("ambiguous", 2) + + +def test_pixel_census_counts_a_copy_when_verify_is_off(tmp_path) -> None: + """Verify disabled never yields 'verified'; the copy must still count.""" + + decision = _pixel_census(tmp_path, _two_copies(), verify=False) + assert decision == CensusDecision("ambiguous", 2) + + +@pytest.mark.parametrize( + "verdict", + [None, IdentityCheck(status="abstain", mode="pixel")], + ids=["no-verdict", "abstain"], +) +def test_pixel_census_counts_anything_short_of_a_mismatch( + tmp_path, monkeypatch, verdict +) -> None: + decision = _pixel_census( + tmp_path, _two_copies(), verify=True, monkeypatch=monkeypatch, verdict=verdict + ) + assert decision.verdict == "ambiguous" + + +def test_pixel_census_rules_out_a_clear_mismatch(tmp_path, monkeypatch) -> None: + decision = _pixel_census( + tmp_path, + _two_copies(), + verify=True, + monkeypatch=monkeypatch, + verdict=IdentityCheck(status="mismatch", mode="pixel"), + ) + assert decision == CensusDecision("unique", 1) + + +def test_pixel_census_is_uncounted_at_another_scale(tmp_path) -> None: + """The target's own place is not found at the recorded scale.""" + + import cv2 + + frame = cv2.resize(_two_copies(), None, fx=1.3, fy=1.3) + decision = _pixel_census(tmp_path, frame, verify=True) + assert decision.verdict == "uncounted" + + +def test_pixel_census_is_uncounted_when_a_place_cannot_be_cut( + tmp_path, monkeypatch +) -> None: + from openadapt_flow.runtime import identity as identity_mod + + real = identity_mod.crop_region + # The target's own crop cuts; the second place does not. + monkeypatch.setattr( + identity_mod, + "crop_region", + lambda png, region: real(png, region) if region[1] == 0 else None, + ) + decision = _pixel_census(tmp_path, _two_copies(), verify=True) + assert decision.verdict == "uncounted" + + +# -- pixel census tie-break: sibling tabs share chrome, not their label --------- + +_TAB_W, _TAB_H = 300, 40 +_TABS = ["Who", "Contact", "Choices", "Employer", "Stats", "Misc"] + + +def _tab(label: str) -> Any: + """An OpenEMR-like section toggle: chrome dominates, the label is small.""" + + import cv2 + import numpy as np + + img = np.full((_TAB_H, _TAB_W), 245, np.uint8) + cv2.rectangle(img, (0, 0), (_TAB_W - 1, _TAB_H - 1), 120, 2) + cv2.rectangle(img, (0, 0), (_TAB_W - 1, 10), 200, -1) + cv2.circle(img, (16, 25), 6, 60, -1) + cv2.putText( + img, label, (30, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.45, 40, 1, cv2.LINE_AA + ) + return img + + +def _tabs(labels: list[str]) -> Any: + import numpy as np + + gap = np.full((12, _TAB_W), 255, np.uint8) + return np.vstack([part for label in labels for part in (_tab(label), gap)]) + + +class _TabOcr: + """OCR that reads a tab crop as its nearest rendered label (or nothing).""" + + def __init__(self, readable: bool = True): + self.readable = readable + + def ocr(self, png: bytes, *, region=None): + import cv2 + import numpy as np + + if not self.readable: + return [] + crop = cv2.imdecode(np.frombuffer(png, np.uint8), cv2.IMREAD_GRAYSCALE) + if crop is None or crop.shape != (_TAB_H, _TAB_W): + return [] + label = min( + _TABS + ["Contact"], + key=lambda name: float(np.abs(crop.astype(int) - _tab(name)).mean()), + ) + return [types.SimpleNamespace(text=label, confidence=1.0)] + + +def _tab_census(tmp_path, labels: list[str], target: int, *, readable: bool = True): + (tmp_path / "templates").mkdir(parents=True, exist_ok=True) + recorded = _png(_tab(labels[target])) + (tmp_path / "templates" / "id.png").write_bytes(recorded) + (tmp_path / "templates" / "t.png").write_bytes(recorded) + top = target * (_TAB_H + 12) + step = Step( + id="open_contact", + intent="click 'Contact'", + action=ActionKind.CLICK, + identity_armed=True, + anchor=Anchor( + template="templates/t.png", + region=(0, top, _TAB_W, _TAB_H), + click_point=(0, top), + identifier_crop="templates/id.png", + identifier_region=(0, top, _TAB_W, _TAB_H), + ), + ) + replayer = Replayer( + _PixelBackend(), + vision=_TabOcr(readable), + poll_interval_s=0.01, + pixel_verify_enabled=True, + ) + resolution = Resolution( + rung="template", point=(0, top), confidence=1.0, elapsed_ms=1 + ) + return replayer._pixel_census( + step, resolution, _png(_tabs(labels)), Workflow(name="wf"), tmp_path + ) + + +def test_a_section_tab_among_sibling_tabs_is_unique_by_its_label(tmp_path) -> None: + """The OpenEMR "Contact" toggle: its siblings share the chrome, and some + are not a clear pixel mismatch, but their labels read differently.""" + + assert _tab_census(tmp_path, _TABS, 1) == CensusDecision("unique", 1) + + +def test_sibling_tabs_with_unreadable_labels_still_halt(tmp_path) -> None: + decision = _tab_census(tmp_path, _TABS, 1, readable=False) + assert decision.verdict == "ambiguous" + + +def test_two_tabs_with_the_same_label_halt(tmp_path) -> None: + decision = _tab_census(tmp_path, ["Who", "Contact", "Misc", "Contact"], 1) + assert decision == CensusDecision("ambiguous", 2) + + +def test_featureless_crops_are_recognized() -> None: + import cv2 + import numpy as np + + from openadapt_flow.vision.match import crop_is_featureless + + band = np.full((64, 160), 40, np.uint8) + band[32:34, :] = 220 # OpenEMR step_016: a band with one rule + assert crop_is_featureless(_png(band)) + assert crop_is_featureless(_png(np.full((48, 240), 128, np.uint8))) + text = np.full((48, 240), 255, np.uint8) + cv2.putText(text, "MG4408 Okafor", (6, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.7, 0, 2) + assert not crop_is_featureless(_png(text)) + assert not crop_is_featureless(_png(_tab("Contact"))) + + +class _ScriptedOcr: + """OCR that reads the target crop first, then every other place, as given.""" + + def __init__(self, target: tuple[str, float], others: tuple[str, float]): + self._target = target + self._others = others + self._calls = 0 + + def ocr(self, png: bytes, *, region=None): + self._calls += 1 + text, confidence = self._target if self._calls == 1 else self._others + return [types.SimpleNamespace(text=text, confidence=confidence)] + + +def _identical_tabs_census(tmp_path, ocr) -> CensusDecision: + """Two identical "Contact" tabs, read through ``ocr``.""" + + labels = ["Contact", "Contact"] + (tmp_path / "templates").mkdir(parents=True, exist_ok=True) + recorded = _png(_tab("Contact")) + (tmp_path / "templates" / "id.png").write_bytes(recorded) + (tmp_path / "templates" / "t.png").write_bytes(recorded) + step = Step( + id="open_contact", + intent="click 'Contact'", + action=ActionKind.CLICK, + identity_armed=True, + anchor=Anchor( + template="templates/t.png", + region=(0, 0, _TAB_W, _TAB_H), + click_point=(0, 0), + identifier_crop="templates/id.png", + identifier_region=(0, 0, _TAB_W, _TAB_H), + ), + ) + replayer = Replayer( + _PixelBackend(), vision=ocr, poll_interval_s=0.01, pixel_verify_enabled=True + ) + resolution = Resolution(rung="template", point=(0, 0), confidence=1.0, elapsed_ms=1) + return replayer._pixel_census( + step, resolution, _png(_tabs(labels)), Workflow(name="wf"), tmp_path + ) + + +def test_a_low_confidence_misread_never_tells_identical_rows_apart(tmp_path) -> None: + """JPEG noise read "Smith, John" as "Smith, 10 lohn" with low confidence: + that reading is unreadable, not a different label.""" + + ocr = _ScriptedOcr(("Smith, John", 0.97), ("Smith, 10 lohn", 0.55)) + assert _identical_tabs_census(tmp_path, ocr) == CensusDecision("ambiguous", 2) + + +def test_one_label_read_two_ways_is_still_the_same_label(tmp_path) -> None: + """A confident misread close to the target text ("Smith, S John") or one + that contains it is the same label, not a sibling.""" + + for other in ("Smith, S John", "Smith, John Jr", "John"): + ocr = _ScriptedOcr(("Smith, John", 0.97), (other, 0.95)) + assert _identical_tabs_census(tmp_path, ocr).verdict == "ambiguous", other + + +def test_an_edge_label_in_a_wide_crop_is_not_featureless() -> None: + """A short label at one edge of a wide cell ("Li, Al" in 200 px) leaves one + shifted window flat; that is structure, not a blank band.""" + + import cv2 + import numpy as np + + from openadapt_flow.vision.match import crop_is_featureless + + cell = np.full((32, 200), 255, np.uint8) + cv2.putText( + cell, "Li, Al", (2, 22), cv2.FONT_HERSHEY_SIMPLEX, 0.5, 0, 1, cv2.LINE_AA + ) + assert not crop_is_featureless(_png(cell)) diff --git a/packages/flow/tests/test_realism_browser.py b/packages/flow/tests/test_realism_browser.py index 1f8ac623d..664bb8e65 100644 --- a/packages/flow/tests/test_realism_browser.py +++ b/packages/flow/tests/test_realism_browser.py @@ -455,6 +455,45 @@ def test_a_backend_that_cannot_count_rows_still_runs_a_plain_form( page.close() +def test_browser_record_ids_follow_the_same_positional_rule(browser) -> None: + """The page-side record-id filter drops exactly what positional_id drops.""" + + import uuid + + from openadapt_flow.backends.playwright_backend import _RECORD_KEY_JS + from openadapt_flow.runtime.identity_census import positional_id + + values = [ + str(uuid.uuid4()), + uuid.uuid4().hex, + f"encounter-{uuid.uuid4()}", + "MRN-A1234", + "00012345", + "chart-1001", + "ENC2024-000123", + "01HEHRGVBY0HPFKMYZV5Q9KR35", + "row-3", + "r3c2", + "item_12", + "row-3-col-1", + "3-1", + "ListViewItem-0", + "2", + ] + page = browser.new_page() + try: + items = "".join(f'
  • x
    • ' for v in values) + page.set_content(f"") + keys = page.evaluate( + f"() => Array.from(document.querySelectorAll('li'))" + f".map((li) => ({_RECORD_KEY_JS})(li, li))" + ) + finally: + page.close() + for index, (value, key) in enumerate(zip(values, keys)): + assert (key is None) == positional_id(value, index), (value, key) + + def _row_buttons(query: dict[str, list[str]]) -> str: """Identical table rows, each with an Edit button whose id is positional.""" order = query.get("order", ["ab"])[0] diff --git a/packages/flow/tests/test_risk.py b/packages/flow/tests/test_risk.py index 070dec4c8..61001c14a 100644 --- a/packages/flow/tests/test_risk.py +++ b/packages/flow/tests/test_risk.py @@ -61,6 +61,13 @@ class TestHeuristic: "Approve", "Pay now", "Sign up", + "Sign visit note", + "Co-sign", + "Sign out patient", + "Sign-out patient Jane Doe", + "Log out patient", + "Discharge patient", + "Check out patient", "Next", "Continue", "Cancel", @@ -74,6 +81,15 @@ def test_write_shaped_true(self, text: str) -> None: [ "Login", "Sign In", + "Sign in with SSO", + "Sign on", + "Sign out", + "Sign Out", + "Signing out", + "Sign out of OpenAdapt", + "Sign out now", + "Sign out all devices", + "Log out", "Search", "Belford, Phil", "Address book", # 'add' must not trip inside 'address' @@ -85,6 +101,17 @@ def test_write_shaped_true(self, text: str) -> None: def test_write_shaped_false(self, text: str) -> None: assert not is_write_shaped(text) + def test_the_session_sign_out_button_stays_reversible(self) -> None: + """Intent and label both read "Sign out": the labels must not join + into "sign out Sign" and look like a clinical sign-out.""" + + step = _click_step("click 'Sign out'", "Sign out") + assert classify_step_risk(step) == "reversible" + + def test_signing_out_a_patient_is_irreversible(self) -> None: + step = _click_step("click 'Sign out patient'", "Sign out patient") + assert classify_step_risk(step) == "irreversible" + def test_submit_click_is_irreversible(self) -> None: assert ( classify_step_risk(_click_step("click 'Save as new message'", "Save"))