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