| """What changed between two observations - in elements, not pixels. |
| |
| The harness already refuses to guess and insists on evidence. This is the missing half of |
| that habit: a structural way to say *what changed* on a screen, so a verifier can be written |
| without hard-coding a string that only today's dialog happens to contain, and so "nothing |
| changed" is a measurement rather than a hunch. |
| |
| before = screen.capture() |
| ... an action ... |
| delta = diff(before, screen.capture()) |
| if delta.closed: # the dialog went away |
| |
| verify = ChangeVerifier(disappeared=["Save As"]) # AgentLoop(verify=verify) |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
| from math import hypot |
| from typing import Any, Iterable |
|
|
| from .observe import Element, Observation, normalise_label |
|
|
| DEFAULT_TOLERANCE = 3 |
| """Pixels. Below this a box has not really moved, it has been re-rendered.""" |
|
|
|
|
| def element_key(element: Element) -> str: |
| """Identity for matching across two frames: the label as the harness compares it, plus kind.""" |
| return f"{normalise_label(element.label)}|{element.kind}" |
|
|
|
|
| def _distance(left: Element, right: Element) -> float: |
| return hypot(left.center[0] - right.center[0], left.center[1] - right.center[1]) |
|
|
|
|
| @dataclass(frozen=True) |
| class Delta: |
| """What a frame gained, lost, or moved.""" |
|
|
| added: tuple[Element, ...] = () |
| removed: tuple[Element, ...] = () |
| moved: tuple[tuple[Element, Element], ...] = () |
|
|
| @property |
| def changed(self) -> bool: |
| """True when anything at all differs.""" |
| return bool(self.added or self.removed or self.moved) |
|
|
| @property |
| def empty(self) -> bool: |
| return not self.changed |
|
|
| @property |
| def closed(self) -> bool: |
| """Something disappeared and nothing appeared: a dialog, menu or panel went away.""" |
| return bool(self.removed) and not self.added |
|
|
| @property |
| def opened(self) -> bool: |
| """Something appeared: a menu, a dialog, a new row.""" |
| return bool(self.added) |
|
|
| @property |
| def settled(self) -> bool: |
| """Nothing moved without also appearing or disappearing - a redraw, not a reaction.""" |
| return bool(self.moved) and not self.added and not self.removed |
|
|
| def labels(self) -> tuple[tuple[str, str], ...]: |
| """(change, label) pairs, for a table or a log line.""" |
| rows = [("added", e.label) for e in self.added] |
| rows += [("removed", e.label) for e in self.removed] |
| rows += [("moved", after.label) for _, after in self.moved] |
| return tuple(rows) |
|
|
| def summary(self) -> str: |
| if self.empty: |
| return "no change" |
| parts = [] |
| if self.added: |
| parts.append(f"+{len(self.added)} ({', '.join(e.label for e in self.added[:4])})") |
| if self.removed: |
| parts.append(f"-{len(self.removed)} ({', '.join(e.label for e in self.removed[:4])})") |
| if self.moved: |
| parts.append(f"~{len(self.moved)}") |
| return " ".join(parts) |
|
|
| def matches(self, *, appeared: Iterable[str] = (), disappeared: Iterable[str] = ()) -> bool: |
| """Does this delta contain the change a caller asked for? Labels compared as the index does.""" |
| def seen(wanted: Iterable[str], pool: Iterable[Element]) -> bool: |
| labels = [normalise_label(e.label) for e in pool] |
| for needle in wanted: |
| target = normalise_label(needle) |
| if not any(target in label or label in target for label in labels): |
| return False |
| return True |
|
|
| return seen(appeared, self.added) and seen(disappeared, self.removed) |
|
|
| def to_dict(self) -> dict[str, Any]: |
| return {"added": [e.label for e in self.added], |
| "removed": [e.label for e in self.removed], |
| "moved": [after.label for _, after in self.moved]} |
|
|
|
|
| def diff(before: Observation, after: Observation, *, tolerance: int = DEFAULT_TOLERANCE) -> Delta: |
| """Match elements by label and kind, and report what was added, removed or moved.""" |
| def group(observation: Observation) -> dict[str, list[Element]]: |
| buckets: dict[str, list[Element]] = {} |
| for element in observation.elements: |
| buckets.setdefault(element_key(element), []).append(element) |
| return buckets |
|
|
| old, new = group(before), group(after) |
| added: list[Element] = [] |
| removed: list[Element] = [] |
| moved: list[tuple[Element, Element]] = [] |
| for key in set(old) | set(new): |
| olds, news = list(old.get(key, ())), list(new.get(key, ())) |
| for left, right in zip(olds, news): |
| if _distance(left, right) > tolerance: |
| moved.append((left, right)) |
| removed.extend(olds[len(news):]) |
| added.extend(news[len(olds):]) |
| return Delta(tuple(added), tuple(removed), tuple(moved)) |
|
|
|
|
| class ChangeVerifier: |
| """A verifier that remembers the previous frame. |
| |
| Pass it straight to ``AgentLoop(verify=...)``: it is called with each new observation and |
| answers whether the change it was asked for happened. Because it keeps the frame it last |
| saw, "the dialog closed" is checkable without a string that depends on the dialog's text. |
| """ |
|
|
| def __init__(self, *, appeared: Iterable[str] = (), disappeared: Iterable[str] = (), |
| tolerance: int = DEFAULT_TOLERANCE) -> None: |
| self.appeared = tuple(appeared) |
| self.disappeared = tuple(disappeared) |
| self.tolerance = tolerance |
| self.previous: Observation | None = None |
| self.last: Delta | None = None |
|
|
| def __call__(self, observation: Observation) -> bool: |
| if self.previous is None: |
| self.previous, self.last = observation, Delta() |
| return False |
| self.last = diff(self.previous, observation, tolerance=self.tolerance) |
| self.previous = observation |
| return self.last.matches(appeared=self.appeared, disappeared=self.disappeared) |
|
|
| def reset(self) -> None: |
| self.previous = None |
| self.last = None |
|
|
