|
| 1 | +"""Attribute a screen change to the specific elements that changed. |
| 2 | +
|
| 3 | +The existing diffs answer "*where* did pixels change" (``motion_regions``, |
| 4 | +``perceptual_diff``, ``ssim_changed_regions`` return raw pixel regions) or "which |
| 5 | +*accessibility* elements differ" (``element_diff``, needs a11y metadata). The |
| 6 | +missing middle is: given a frame diff **and a list of element boxes**, which of |
| 7 | +*those* elements changed? ``change_localize`` scores each supplied box by how |
| 8 | +much it changed and ranks them. |
| 9 | +
|
| 10 | +* :func:`rank_changes` — pure: take ``[{box, score}]`` and mark each box |
| 11 | + ``changed`` (score at or above ``threshold``), sorted most-changed first. |
| 12 | +* :func:`localize_changes` — diff a reference against the current screen, score |
| 13 | + each element box by its mean pixel change, and rank them. |
| 14 | +
|
| 15 | +cv2 / numpy are imported lazily (the module stays importable without them) and |
| 16 | +the loaders reuse :mod:`visual_match`. The ranking is pure and fully testable. |
| 17 | +Imports no ``PySide6``. |
| 18 | +""" |
| 19 | +from typing import Any, Dict, List, Optional, Sequence |
| 20 | + |
| 21 | + |
| 22 | +def _unpack(item: Any) -> tuple: |
| 23 | + """Return ``(box, score)`` from a ``{box, score}`` dict or a ``(box, score)``.""" |
| 24 | + if isinstance(item, dict): |
| 25 | + return item["box"], item["score"] |
| 26 | + return item[0], item[1] |
| 27 | + |
| 28 | + |
| 29 | +def rank_changes(scored_boxes: Sequence[Any], *, |
| 30 | + threshold: float = 0.1) -> List[Dict[str, Any]]: |
| 31 | + """Mark and rank scored element boxes by how much they changed (pure). |
| 32 | +
|
| 33 | + ``scored_boxes`` is a sequence of ``{box, score}`` (or ``(box, score)``). |
| 34 | + Returns ``[{box, score, changed}]`` sorted by descending score; ``changed`` |
| 35 | + is ``True`` when the score is at or above ``threshold``. |
| 36 | + """ |
| 37 | + limit = float(threshold) |
| 38 | + result = [ |
| 39 | + {"box": [int(value) for value in box], |
| 40 | + "score": round(float(score), 4), |
| 41 | + "changed": float(score) >= limit} |
| 42 | + for box, score in (_unpack(item) for item in scored_boxes) |
| 43 | + ] |
| 44 | + result.sort(key=lambda entry: entry["score"], reverse=True) |
| 45 | + return result |
| 46 | + |
| 47 | + |
| 48 | +def _box_mean(diff: Any, box: Sequence[int]) -> float: |
| 49 | + """Mean change (0..1) of the diff map inside ``box`` (numpy).""" |
| 50 | + x, y, w, h = (int(box[0]), int(box[1]), int(box[2]), int(box[3])) |
| 51 | + patch = diff[max(0, y):y + h, max(0, x):x + w] |
| 52 | + return float(patch.mean()) if patch.size else 0.0 |
| 53 | + |
| 54 | + |
| 55 | +def localize_changes(reference: Any, boxes: Sequence[Sequence[int]], *, |
| 56 | + current: Optional[Any] = None, threshold: float = 0.1, |
| 57 | + region: Optional[Sequence[int]] = None |
| 58 | + ) -> List[Dict[str, Any]]: |
| 59 | + """Score and rank which of ``boxes`` changed between two frames. |
| 60 | +
|
| 61 | + Diffs ``reference`` against ``current`` (a fresh screen grab of ``region`` |
| 62 | + by default), takes each box's mean per-pixel change (0..1), and ranks them |
| 63 | + via :func:`rank_changes`. Returns ``[{box, score, changed}]``. |
| 64 | + """ |
| 65 | + import numpy as np |
| 66 | + from je_auto_control.utils.visual_match.visual_match import ( |
| 67 | + _grab_gray, _to_gray) |
| 68 | + ref = _to_gray(reference).astype("float64") |
| 69 | + other = current if current is not None else _grab_gray(region) |
| 70 | + cur = _to_gray(other).astype("float64") |
| 71 | + diff = np.abs(ref - cur) / 255.0 |
| 72 | + scored = [{"box": list(box), "score": _box_mean(diff, box)} |
| 73 | + for box in boxes] |
| 74 | + return rank_changes(scored, threshold=threshold) |
0 commit comments