File size: 11,083 Bytes
59022ca
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
61aac99
 
 
 
 
 
 
 
 
59022ca
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
b038669
 
 
59022ca
61aac99
 
 
 
 
 
 
59022ca
b038669
 
 
 
 
 
61aac99
b038669
61aac99
59022ca
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
61aac99
b038669
 
61aac99
b038669
 
 
61aac99
 
 
 
 
b038669
59022ca
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
b038669
 
 
 
61aac99
 
 
 
b038669
 
 
61aac99
 
 
 
 
 
 
b038669
59022ca
 
 
 
 
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
"""Turning pixels into a labelled element index.

This is the module that makes a *tiny* model viable. The model never has to guess
where anything is, because OCR has already turned the screen into a short list of
labels with coordinates - and the model only has to pick one.

Filters matter as much as the engine. Raw OCR returns dialog prose, fragmentary
debris and stray punctuation, and every extra line of noise is another chance for a
small model to pick the wrong element, so the index is cleaned aggressively.

Everything here is optional at import time: a missing OCR engine raises a clear
:class:`DetectorUnavailable` instead of breaking the harness.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Protocol

from tiny_computer_agent.observe import Element, Observation, ScreenSource


class DetectorUnavailable(RuntimeError):
    """Raised when an element detector is requested but cannot run on this machine."""


class Detector(Protocol):
    """Turns one frame into labelled, clickable elements."""

    name: str

    def detect(self, image: Any) -> tuple[Element, ...]: ...


@dataclass
class OcrDetector:
    """Indexes on-screen text with RapidOCR: ONNX, CPU, offline after install.

    ``max_label_chars`` is the important knob. Screen text includes paragraphs and
    titles that are not widgets, and a model shown them will happily click prose -
    which is exactly what a 0.5B planner did when dialog text sat in its menu. Short
    strings are much more likely to be buttons and menu items, so the default is
    deliberately tight.
    """

    min_score: float = 0.6
    max_label_chars: int = 24
    min_label_chars: int = 2
    name: str = "ocr"
    _engine: Any = field(default=None, init=False, repr=False)

    def _loaded(self) -> Any:
        if self._engine is None:
            try:
                from rapidocr_onnxruntime import RapidOCR
            except ImportError as exc:  # pragma: no cover - depends on the machine
                raise DetectorUnavailable(
                    "no OCR engine available: pip install 'tiny-computer-agent[ocr]'"
                ) from exc
            self._engine = RapidOCR()
        return self._engine

    def detect(self, image: Any) -> tuple[Element, ...]:
        import numpy as np

        array = np.asarray(image.convert("RGB") if hasattr(image, "convert") else image)
        result, _ = self._loaded()(array)
        elements: list[Element] = []
        for box, text, score in result or []:
            element = self._to_element(box, text, float(score))
            if element is not None:
                elements.append(element)
        return tuple(elements)

    def _to_element(self, box: Any, text: Any, score: float) -> Element | None:
        """Convert one OCR box to an Element, or None if it is not a usable label."""
        label = " ".join(str(text).split())
        if score < self.min_score:
            return None
        if not self.min_label_chars <= len(label) <= self.max_label_chars:
            return None
        if not any(character.isalnum() for character in label):
            return None
        xs = [float(point[0]) for point in box]
        ys = [float(point[1]) for point in box]
        x, y = int(min(xs)), int(min(ys))
        return Element(
            label=label, kind="text",
            x=x, y=y,
            width=int(max(xs)) - x, height=int(max(ys)) - y,
        )


@dataclass
class IndexedScreen:
    """Composes pixel capture with a detector, so ``capture()`` yields an index.

    This is the wiring that turns the harness from a scripted demo into something
    that can act on a real desktop.
    """

    source: ScreenSource
    detector: Detector
    fill_text: bool = True

    def capture(self) -> Observation:
        observation = self.source.capture()
        try:
            elements = self.detector.detect(observation.frame)
        except DetectorUnavailable:
            raise
        except Exception as exc:
            # A perception failure must not kill a run that is otherwise fine.
            elements = ()
            observation = Observation(
                index=observation.index, frame=observation.frame,
                elements=(), text=f"perception error: {type(exc).__name__}: {exc}",
            )
            return observation
        mapper = getattr(self.source, "map_to_screen", None)
        # Which space did *this call* produce? A detector that chains several
        # sources (the "best" detector: accessibility tree, then OCR, then grid)
        # cannot answer that with a static flag, and answering it wrongly shifts
        # every click by the capture's scale factor.
        space = "frame"
        hook = getattr(self.detector, "coordinates_of_last_detect", None)
        if callable(hook):
            space = hook()
        if mapper is not None and space != "screen":
            # Frame pixels -> absolute screen pixels, so a click cannot drift.
            elements = tuple(mapper(element) for element in elements)
        text = observation.text
        if self.fill_text and not text:
            # Grid labels are coordinates, not visible text, so they are not page text.
            text = " ".join(element.label for element in elements if element.kind != "grid")
        return Observation(
            index=observation.index, frame=observation.frame, elements=elements, text=text,
        )


@dataclass
class GridDetector:
    """A numbered grid over the frame: click targets that need nothing installed.

    The guarantee that matters. OCR finds text, the accessibility tree finds widgets,
    but a canvas app, a game, a photo, a custom-drawn toolbar - anything with no text
    - leaves a text index empty, and an agent with an empty index has *no legal
    moves*. A grid always has moves, and a numbered menu is something even a 256M
    model can choose from reliably.
    """

    rows: int = 8
    cols: int = 12
    name: str = "grid"

    def __post_init__(self) -> None:
        if self.rows < 1 or self.cols < 1:
            raise ValueError("rows and cols must be at least 1")

    def detect(self, image: Any) -> tuple[Element, ...]:
        width, height = image.size
        cell_w = width / self.cols
        cell_h = height / self.rows
        elements = []
        for row in range(self.rows):
            for col in range(self.cols):
                elements.append(Element(
                    label=f"cell {row * self.cols + col + 1}",
                    kind="grid",
                    x=round(col * cell_w), y=round(row * cell_h),
                    width=max(1, round(cell_w)), height=max(1, round(cell_h)),
                ))
        return tuple(elements)


@dataclass
class AutoDetector:
    """OCR when it finds text, numbered cells when it does not.

    This is the default, so the harness is useful on a machine with nothing installed
    beyond Pillow - and gets better automatically once an OCR engine appears.
    """

    name: str = "auto"
    ocr: Detector = field(default_factory=OcrDetector)
    grid: Detector = field(default_factory=GridDetector)
    # opt in (the "best" detector does): the accessibility tree is tried first,
    # because it is exact where OCR is a guess - and skipped when it is empty
    accessibility: Detector | None = None

    def __post_init__(self) -> None:
        self._last_space = "frame"

    def coordinates_of_last_detect(self) -> str:
        """Whatever the branch that actually produced the last index spoke."""
        return self._last_space

    def detect(self, image: Any) -> tuple[Element, ...]:
        if self.accessibility is not None:
            try:
                elements = self.accessibility.detect(image)
            except DetectorUnavailable:
                elements = ()
            if elements:
                self._last_space = "screen"
                return elements
        self._last_space = "frame"
        try:
            elements = self.ocr.detect(image)
        except DetectorUnavailable:
            return self.grid.detect(image)
        return elements or self.grid.detect(image)


def _ocr_available() -> tuple[bool, str]:
    try:
        import rapidocr_onnxruntime  # noqa: F401
    except ImportError:
        return False, "rapidocr-onnxruntime is not installed (pip install 'tiny-computer-agent[ocr]')"
    try:
        import onnxruntime  # noqa: F401
    except ImportError:  # pragma: no cover
        return False, "onnxruntime is not installed"
    return True, "rapidocr-onnxruntime ready (CPU, offline)"


def detector_available(name: str = "auto") -> tuple[bool, str]:
    """(available, detail) for preflight and the ``doctor`` command."""
    if name == "grid":
        return True, "numbered grid cells (no dependencies)"
    if name in ("atspi", "uia"):
        from tiny_computer_agent.accessibility import accessibility_available

        return accessibility_available(name)
    if name == "best":
        from tiny_computer_agent.accessibility import accessibility_available

        for backend in ("atspi", "uia"):            # Linux first, then Windows
            reachable, detail = accessibility_available(backend)
            if reachable:
                return True, (f"{backend} tree when it has content, then OCR, then a "
                              f"numbered grid ({detail})")
        return True, (f"OCR, then a numbered grid (no accessibility tree: {detail})")
    if name == "auto":
        available, detail = _ocr_available()
        if available:
            return True, f"OCR when it finds text, grid otherwise ({detail})"
        return True, "grid only: no OCR engine installed (pip install 'tiny-computer-agent[ocr]')"
    if name != "ocr":
        return False, f"unknown detector {name!r}"
    return _ocr_available()


def make_detector(name: str = "auto", **kwargs: Any) -> Detector:
    """Build a detector by name, with an actionable error when it cannot run."""
    available, detail = detector_available(name)
    if not available:
        raise DetectorUnavailable(detail)
    if name == "grid":
        return GridDetector(**kwargs)
    if name == "atspi":
        from tiny_computer_agent.accessibility import AtspiDetector

        return AtspiDetector(**kwargs)
    if name == "uia":
        from tiny_computer_agent.accessibility import UiaDetector

        return UiaDetector(**kwargs)
    if name == "best":
        from tiny_computer_agent.accessibility import AtspiDetector, accessibility_available

        from tiny_computer_agent.accessibility import AtspiDetector, UiaDetector

        accessibility = None
        if accessibility_available("atspi")[0]:
            accessibility = AtspiDetector()
        elif accessibility_available("uia")[0]:
            accessibility = UiaDetector()
        return AutoDetector(accessibility=accessibility, **kwargs)
    if name == "auto":
        if _ocr_available()[0]:
            return AutoDetector(**kwargs)
        return GridDetector(**kwargs)
    return OcrDetector(**kwargs)