File size: 6,072 Bytes
e2f048d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """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
|