Multi-monitor jumps, screen OCR, and generated sprite art

petctl gains screen verbs: `jump` (1-based number, name, next/prev/
primary/other, or a direction resolved from real geometry), `monitors`,
and `read` for OCR of a monitor's contents.

- monitors.py: pure layout model + jump-target resolution. The monitor
  list is published by PetWindow from QGuiApplication.screens() over a
  queued signal, so the controller and window agree on what "monitor 2"
  means; xrandr and Qt order screens differently on the same machine.
- screen_text.py: pull-only OCR (mss capture + Tesseract/RapidOCR).
  Nothing captures unless the server asks, and the text rides back up
  the tool-result relay so Bolt can read a screen mid-turn. Both deps
  optional, soft-failing with a reason. SCREEN_TEXT=false removes it.
- Query verbs are answered in controller._handle_command rather than
  pet_actions.describe(), because their output is the point.
- scripts/generate_bolt_sprites.py draws every frame; walk/ is a
  side-view cycle stepped by distance travelled, not by the animation
  timer, so the planted paw tracks the window exactly. sprite.py loads
  it via EXTRA_ANIMATIONS keyed by name, with has() so callers can
  decline a placeholder blob.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-28 16:17:11 -06:00
parent ccb3aeb7ae
commit b121bbba17
50 changed files with 2460 additions and 49 deletions
+64 -17
View File
@@ -1,37 +1,84 @@
# Sprite assets
Art: [Kenney's Robot Pack](https://kenney.nl/assets/robot-pack) (CC0 — no
attribution required, credited here anyway), the green side-view robot.
Source pack lives at `~/Documents/kenney_robot-pack`; only the frames listed
below were copied in.
Art: Bolt himself — a cream shepherd pup with a slate cap, a lightning blaze
on his forehead and a bolt tag on his collar. The frames are **generated, not
hand-drawn**: `scripts/generate_bolt_sprites.py` draws every one of them with
Pillow and writes this folder.
```bash
python scripts/generate_bolt_sprites.py # rewrite this folder
python scripts/generate_bolt_sprites.py --out /tmp/prev # preview elsewhere first
python scripts/generate_bolt_sprites.py --states idle # just one state
```
That means tweaking the art is editing code, not 24 PNGs: the palette is a
block of constants at the top of the script, the body/head/ear/tail shapes are
one function each in normalised 0..1 coordinates, and each state's animation is
a list of pose dicts in `frames_for()`. Everything is super-sampled 4x and
downscaled on save, because PIL's draw primitives have no antialiasing.
**Regenerate after editing** — the PNGs here are committed, so a change to the
script alone doesn't move the pet.
Convention the loader (`bolt_pet/ui/sprite.py`) expects:
```
assets/sprites/
idle/ frame_00.png robot_greenBody (standing)
listening/ frame_00.png, frame_01.png robot_greenDrive1/2 (tracks rolling — "leaning in")
thinking/ frame_00.png, frame_01.png robot_greenDamage1/2 (flicker — "processing")
talking/ frame_00.png, frame_01.png robot_greenBody, robot_greenJump (bounce)
error/ frame_00.png robot_greenHurt
idle/ frame_00..07.png breathing, tail wag, blink on frame 06
listening/ frame_00..03.png ears perked, head tilted in, collar tag lit, sound arcs
thinking/ frame_00..05.png eyes up, head cocked, cycling dots
talking/ frame_00..03.png mouth open/close with tongue, ears bouncing
error/ frame_00..01.png X eyes, ears drooped, red spark
walk/ frame_00..07.png side-view walk cycle (see below)
```
- One subfolder per pet state (matches `bolt_pet.state.PetState`).
- One subfolder per pet state (matches `bolt_pet.state.PetState`), **plus
`walk/`**, which is not a state — see below.
- Any `*.png` filenames work — they're played back in alphabetical-sort
order, looping, at `IDLE_ANIMATION_FPS` (see `.env`).
- Frames are scaled to fit within `PET_SIZE` (default 160px), keeping aspect
ratio, and centered in the (square) pet window — the source art here isn't
square, so don't assume it fills the frame edge-to-edge.
order, looping, at `IDLE_ANIMATION_FPS` (see `.env`). At the default 6fps
the 8-frame idle loop runs about 1.3s.
- Frames are square (320px, 2x the default `PET_SIZE` of 160) so they
downscale cleanly; the loader scales to fit `PET_SIZE` keeping aspect ratio
and centres them in the square pet window.
- A state directory with no frames in it falls back to a small
procedurally-drawn placeholder blob (see `_placeholder_frames` in
`sprite.py`).
## The walk cycle
`walk/` is the one animation that isn't a `PetState`. Walking is a property of
*movement* — orthogonal to whether he's idle, listening or talking — so it
stays out of the state machine and is keyed by name instead
(`sprite.EXTRA_ANIMATIONS`). `PetWindow` uses it whenever the pet is actually
travelling and falls back to the state animation the moment it stops.
Three things about it are load-bearing if you redraw it:
- **It's a side view, drawn facing right.** The other poses are a
front-facing sit, which is fine standing still but slides like a chess
piece when moving. `PetWindow._oriented()` mirrors the frames (cached) when
he walks left, so only the right-facing version exists on disk.
- **The cycle is advanced by distance travelled, not by the animation
timer** (`_WALK_PIXELS_PER_FRAME`, one frame per ~13px). That's what keeps
a planted paw tracking backwards at exactly the speed the window moves
forwards. Drive it off the clock and the feet skate whenever
`PET_WANDER_SPEED` doesn't happen to match `IDLE_ANIMATION_FPS`. If you
change the number of frames or the stride length in
`paw_position()`, retune that constant to match or he'll moonwalk.
- **The frames carry their own vertical bob**, so the window's own bob is
switched off while they're in use. Only the no-walk-art fallback still
bobs in code.
Delete `walk/` and everything still runs — he reverts to sliding with a small
coded bob, which is what the pet did before the cycle existed.
## Swapping in different art
Replace any state's PNGs (same alphabetical-order-loops convention) to
change its look — no code changes needed. If your source is a single grid
spritesheet (rows/cols of frames in one PNG) rather than one-file-per-frame,
use `scripts/slice_spritesheet.py` to cut it into this folder-of-frames
change its look — no code changes needed, and nothing forces you to keep
using the generator. If your source is a single grid spritesheet (rows/cols
of frames in one PNG) rather than one-file-per-frame, use
`scripts/slice_spritesheet.py` to cut it into this folder-of-frames
convention:
```bash
Binary file not shown.

Before

Width:  |  Height:  |  Size: 5.6 KiB

After

Width:  |  Height:  |  Size: 55 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 54 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.4 KiB

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 57 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 59 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 56 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.3 KiB

After

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.3 KiB

After

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.4 KiB

After

Width:  |  Height:  |  Size: 59 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 4.0 KiB

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.7 KiB

After

Width:  |  Height:  |  Size: 62 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 2.0 KiB

After

Width:  |  Height:  |  Size: 62 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 62 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 62 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 61 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 61 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 46 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 46 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 43 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 43 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 46 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 45 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 42 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 43 KiB

+24
View File
@@ -144,6 +144,30 @@ TTS_STREAMING = os.environ.get("TTS_STREAMING", "true").lower() in ("1", "true",
SCREEN_CONTEXT = os.environ.get("SCREEN_CONTEXT", "true").lower() in ("1", "true", "yes", "on")
# ── monitors ────────────────────────────────────────────────────────────────
# A one-line note about the screen layout (how many, their sizes, which one
# the pet is standing on) rides along with each utterance, so Bolt can decide
# to `petctl jump` somewhere without asking you what you've got plugged in.
# Cheap — the list comes from the UI, nothing is probed per turn.
MONITOR_CONTEXT = os.environ.get("MONITOR_CONTEXT", "true").lower() in (
"1", "true", "yes", "on"
)
# ── screen text (OCR) ───────────────────────────────────────────────────────
# Lets Bolt actually read a monitor, via `petctl read`. Pull-only: nothing is
# captured unless the server asks for it, and every read is logged. Needs the
# optional capture/OCR extras — see the comments in requirements.txt.
#
# This widens what can leave the machine more than any other switch here: the
# recognised text of a whole screen goes to the server. It is *not* a new
# capability (the shell relay could already run a screenshot tool and OCR it),
# but it is a much easier one to use by accident. Set SCREEN_TEXT=false to
# take it away entirely.
SCREEN_TEXT = os.environ.get("SCREEN_TEXT", "true").lower() in ("1", "true", "yes", "on")
SCREEN_TEXT_MAX_CHARS = int(os.environ.get("SCREEN_TEXT_MAX_CHARS", "4000"))
# ── quiet hours / do-not-disturb ────────────────────────────────────────────
# Comma-separated HH:MM-HH:MM ranges (wrapping midnight is fine). While
# napping the pet dims, stops wandering, and makes no proactive noise —
+82 -4
View File
@@ -20,8 +20,9 @@ from typing import Optional
from PySide6.QtCore import QObject, Signal
from . import (
config, history as history_mod, notifications, pet_actions, quiet,
screen_context, server_client, speech_text, updater,
config, history as history_mod, monitors as monitors_mod, notifications,
pet_actions, quiet, screen_context, screen_text, server_client,
speech_text, updater,
)
from .audio import barge_in, mic, stt, tts, wake_word
from .state import PetState, PetStateMachine
@@ -53,6 +54,13 @@ class PetController(QObject):
# window. Append-only from this thread; the UI only ever snapshots it.
self.history = history_mod.ConversationHistory(limit=config.HISTORY_LIMIT)
# The screen layout, as published by the UI (see set_monitors). Held
# here rather than probed, so "monitor 2" means the same thing to the
# controller and to the window that has to jump there — see
# monitors.py for why that matters.
self._monitors: list[monitors_mod.Monitor] = []
self._pet_monitor: Optional[int] = None
# Wake-word sensitivity is live-tunable (tray tuner), so it's read
# through a callable on every frame rather than captured per listen.
self._wake_threshold = config.WAKE_WORD_THRESHOLD
@@ -243,7 +251,7 @@ class PetController(QObject):
# What's focused right now rides along, so "what's this error?"
# has a referent without you having to describe the window.
reply = server_client.converse(
screen_context.context_for(text), on_command=self._handle_command
self._with_context(text), on_command=self._handle_command
)
except server_client.ServerError as exc:
self.log.emit(f"Server error: {exc}")
@@ -254,6 +262,31 @@ class PetController(QObject):
self._speak(reply)
self._state.transition(PetState.IDLE)
def _with_context(self, text: str) -> str:
"""Everything the server gets alongside what you actually said: the
focused window title, and a one-line note about the screen layout so
Bolt knows how many monitors there are and where he's standing
without having to ask. Only the *layout* rides along for free — the
text on those screens costs an OCR pass, so it stays behind
`petctl read`."""
text = screen_context.context_for(text)
if config.MONITOR_CONTEXT:
text = monitors_mod.annotate(text, self._monitors, self._pet_monitor)
return text
# ── screen layout, published by the UI ───────────────────────────────
def set_monitors(self, monitors: list) -> None:
"""Slot: the window telling us what screens exist (queued signal)."""
self._monitors = list(monitors)
self.log.emit(
"Screens: " + (monitors_mod.summary(self._monitors) or "none reported")
)
def set_pet_monitor(self, index: int) -> None:
"""Slot: the window telling us which screen the pet is standing on."""
self._pet_monitor = int(index)
def _handle_command(self, command: str) -> str:
"""Server-relayed command. `petctl ...` drives the pet's body and
never reaches a shell; everything else is a real command, exactly as
@@ -266,11 +299,56 @@ class PetController(QObject):
if action is None:
return server_client.run_local_command(command)
self.log.emit(f"Pet action: {action}")
if action["action"] == "nap":
# Queries answer from here rather than from pet_actions.describe():
# their output *is* the useful part, and it's what the server reads
# back off the tool-result relay.
kind = action["action"]
if kind == "monitors":
return monitors_mod.describe(self._monitors, self._pet_monitor)
if kind == "read":
return self._read_screen(action["target"])
if kind == "jump":
try:
target = monitors_mod.resolve(
self._monitors, action["target"], self._pet_monitor
)
except ValueError as exc:
self.log.emit(f"petctl jump: {exc}")
return f"[pet] {exc}"
# Hand the window a resolved index, so it can't re-resolve the
# spec against a different screen ordering.
self.action.emit({"action": "jump", "monitor": target.index})
return f"[pet] jumped to monitor {target.label}"
if kind == "nap":
self.set_napping(bool(action["enabled"]))
self.action.emit(action)
return pet_actions.describe(action)
def _read_screen(self, target: str) -> str:
"""`petctl read` — OCR a screen and hand the text back to the server."""
if not config.SCREEN_TEXT:
return "[pet] screen reading is disabled (set SCREEN_TEXT=true in .env)"
if not self._monitors:
return "[pet] no monitor information available"
limit = config.SCREEN_TEXT_MAX_CHARS
if target in ("all", "everything", "*"):
self.log.emit(f"Reading all {len(self._monitors)} screens…")
return screen_text.read_monitors(self._monitors, limit)
if target in ("here", "", "this", "current"):
index = self._pet_monitor if self._pet_monitor is not None else 0
monitor = self._monitors[min(index, len(self._monitors) - 1)]
else:
try:
monitor = monitors_mod.resolve(
self._monitors, target, self._pet_monitor
)
except ValueError as exc:
return f"[pet] {exc}"
self.log.emit(f"Reading monitor {monitor.number} ({monitor.name})…")
return screen_text.read_monitor(monitor, limit)
def _speak(self, text: str) -> None:
self._state.transition(PetState.TALKING)
# Bubble gets the markdown stripped but emoji kept (it can't render
+213
View File
@@ -0,0 +1,213 @@
"""Which screens exist, and which one the pet is standing on.
Deliberately free of Qt *and* of any subprocess probing: the monitor list is
published by the UI (`ui/pet_window.py` builds it from
`QGuiApplication.screens()`) and handed to the controller over a queued
signal, the same way every other UI↔controller message travels.
That indirection is the whole point. The pet has to agree with itself about
what "monitor 2" means — if the controller enumerated screens with `xrandr`
while the window jumped using Qt's screen list, the two orderings could
disagree and Bolt would announce one screen and land on another. Making Qt the
single source of truth removes that class of bug, and leaves everything here
pure enough to unit test without a display.
Indices are **1-based in every string a human or the model ever sees**, and
0-based in the list itself. `Monitor.index` is the 0-based one; `.number` is
what gets printed.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable, Optional
# Directional specs understood by resolve(), mapped to a (dx, dy) heading.
_DIRECTIONS = {
"left": (-1, 0),
"right": (1, 0),
"up": (0, -1),
"above": (0, -1),
"down": (0, 1),
"below": (0, 1),
}
@dataclass(frozen=True)
class Monitor:
"""One screen, in the global desktop coordinate space."""
index: int # 0-based position in the published list
name: str
x: int
y: int
width: int
height: int
primary: bool = False
@property
def number(self) -> int:
"""1-based, for anything a person or the model reads."""
return self.index + 1
@property
def right(self) -> int:
return self.x + self.width
@property
def bottom(self) -> int:
return self.y + self.height
@property
def center(self) -> tuple[int, int]:
return self.x + self.width // 2, self.y + self.height // 2
def contains(self, x: int, y: int) -> bool:
return self.x <= x < self.right and self.y <= y < self.bottom
@property
def label(self) -> str:
bits = f"{self.number}: {self.name} {self.width}x{self.height}"
return bits + " (primary)" if self.primary else bits
def monitor_containing(
monitors: Iterable[Monitor], x: int, y: int
) -> Optional[Monitor]:
"""The screen holding point (x, y), or None if it's off every screen."""
for monitor in monitors:
if monitor.contains(x, y):
return monitor
return None
def nearest_monitor(monitors: Iterable[Monitor], x: int, y: int) -> Optional[Monitor]:
"""Screen whose centre is closest to (x, y) — the fallback when a point
lands in the dead space between mismatched screens."""
monitors = list(monitors)
if not monitors:
return None
return min(
monitors,
key=lambda m: (m.center[0] - x) ** 2 + (m.center[1] - y) ** 2,
)
def resolve(
monitors: list[Monitor], spec: str, current: Optional[int] = None
) -> Monitor:
"""Turn a `petctl jump` target into a screen.
Accepts a 1-based number, a name (case-insensitive substring, so "hdmi"
finds "HDMI-0"), `next`/`prev`, `primary`, `other`, or a direction
(`left`/`right`/`up`/`down`) relative to *current*. Raises ValueError with
a message meant to be read by the model, since it goes back as tool
output.
"""
if not monitors:
raise ValueError("no monitors have been reported yet")
spec = (spec or "").strip().lower()
if not spec:
raise ValueError("jump needs a target monitor")
count = len(monitors)
if current is None or not (0 <= current < count):
current = next((m.index for m in monitors if m.primary), 0)
if spec.isdigit():
number = int(spec)
if not (1 <= number <= count):
raise ValueError(
f"there is no monitor {number}; you have {count} "
f"(1-{count})"
)
return monitors[number - 1]
if spec in ("next", "forward"):
return monitors[(current + 1) % count]
if spec in ("prev", "previous", "back"):
return monitors[(current - 1) % count]
if spec == "primary":
return next((m for m in monitors if m.primary), monitors[0])
if spec == "other":
# With two screens "the other one" is unambiguous; with more it's just
# the next one round, which is at least always a *different* screen.
return monitors[(current + 1) % count]
if spec == "random":
# Deterministic-free choice is the caller's business; pick the screen
# furthest from the current one so "random" always visibly moves.
here = monitors[current].center
return max(
monitors,
key=lambda m: (m.center[0] - here[0]) ** 2 + (m.center[1] - here[1]) ** 2,
)
if spec in _DIRECTIONS:
dx, dy = _DIRECTIONS[spec]
here = monitors[current].center
candidates = []
for monitor in monitors:
if monitor.index == current:
continue
ox, oy = monitor.center
along = (ox - here[0]) * dx + (oy - here[1]) * dy
if along <= 0:
continue # not in that direction at all
drift = abs((ox - here[0]) * dy + (oy - here[1]) * dx)
candidates.append((drift, along, monitor))
if not candidates:
raise ValueError(
f"there's no monitor to the {spec} of monitor "
f"{monitors[current].number}"
)
# Prefer the best-aligned screen, then the closest of those.
candidates.sort(key=lambda item: (item[0], item[1]))
return candidates[0][2]
matches = [m for m in monitors if spec in m.name.lower()]
if len(matches) == 1:
return matches[0]
if len(matches) > 1:
raise ValueError(
f"{spec!r} matches several monitors: "
+ ", ".join(m.label for m in matches)
)
raise ValueError(
f"unknown monitor {spec!r}; you have: "
+ "; ".join(m.label for m in monitors)
+ " — or use next/prev/primary/left/right/up/down"
)
def describe(monitors: list[Monitor], current: Optional[int] = None) -> str:
"""Full listing, used as `petctl monitors` output."""
if not monitors:
return "[pet] no monitor information available"
lines = [f"[pet] {len(monitors)} monitor(s):"]
for monitor in monitors:
here = " <- Bolt is here" if monitor.index == current else ""
lines.append(
f" {monitor.label} at +{monitor.x}+{monitor.y}{here}"
)
return "\n".join(lines)
def summary(monitors: list[Monitor], current: Optional[int] = None) -> Optional[str]:
"""One-line version tacked onto each utterance — short on purpose, since
it rides along with every single thing you say."""
if not monitors:
return None
parts = ", ".join(f"{m.number}) {m.name} {m.width}x{m.height}" for m in monitors)
line = f"{len(monitors)} monitors: {parts}"
if current is not None and 0 <= current < len(monitors):
line += f"; Bolt is on {monitors[current].number}"
return line
def annotate(text: str, monitors: list[Monitor], current: Optional[int] = None) -> str:
"""Attach the screen summary as a separate aside, matching the style of
screen_context.annotate() so the model can ignore it when irrelevant."""
text = (text or "").strip()
line = summary(monitors, current)
if not text or not line:
return text
return f"{text}\n\n[{line}]"
+30
View File
@@ -29,8 +29,18 @@ ANCHORS = (
EMOTES = ("wave", "hop", "spin", "nod", "shake", "bounce", "wiggle")
# Where `petctl jump` can be aimed. A bare number (1-based) works too, as does
# any unique part of a monitor's name — resolution lives in monitors.resolve().
MONITOR_SPECS = (
"next", "prev", "primary", "other", "random",
"left", "right", "up", "down",
)
HELP = (
"petctl move <x> <y> | <" + "|".join(ANCHORS) + ">\n"
"petctl jump <monitor number|" + "|".join(MONITOR_SPECS) + "|name>\n"
"petctl monitors\n"
"petctl read [monitor number|here|all]\n"
"petctl emote <" + "|".join(EMOTES) + ">\n"
"petctl say <text>\n"
"petctl wander on|off\n"
@@ -82,6 +92,24 @@ def parse(command: str) -> Optional[dict]:
raise ActionError(f"unknown position {args[0]!r}; try one of: " + ", ".join(ANCHORS))
return {"action": "move", "anchor": anchor}
if verb in ("jump", "monitor", "screen"):
if not args:
raise ActionError(
"jump needs a monitor: a number, a name, or one of "
+ ", ".join(MONITOR_SPECS)
)
# The spec isn't validated here on purpose: which monitors exist is a
# runtime fact this pure module doesn't have. monitors.resolve() does
# it once the published screen list is in hand.
return {"action": "jump", "target": " ".join(args).strip()}
if verb in ("monitors", "screens", "displays"):
return {"action": "monitors"}
if verb in ("read", "look", "ocr", "see"):
target = (" ".join(args).strip() or "here").lower()
return {"action": "read", "target": target}
if verb in ("emote", "do"):
if not args:
raise ActionError("emote needs a name: " + ", ".join(EMOTES))
@@ -124,6 +152,8 @@ def describe(action: dict) -> str:
if kind == "move":
where = action.get("anchor") or f"({action.get('x')}, {action.get('y')})"
return f"[pet] walking to {where}"
if kind == "jump":
return f"[pet] jumping to monitor {action['target']}"
if kind == "emote":
return f"[pet] {action['emote']}"
if kind == "say":
+187
View File
@@ -0,0 +1,187 @@
"""Reading the text that's actually on a monitor, via screenshot + OCR.
This is the "Bolt can see what's on screen" half of the screen features. It is
**pull, not push**: nothing here runs on its own. The server has to ask, by
relaying `petctl read`, and the recognised text goes back as that command's
output through the existing tool-result relay (see server_client.converse).
That's deliberate on two counts — OCR of a 4K screen costs a second or two,
which would be tacked onto every single utterance if it ran automatically, and
"screen contents leave this machine" should be a thing Bolt decides to do and
you can see in the log, not a silent constant.
Both halves are optional and soft-fail with a reason, the way hotkey.py does:
capture needs `mss`, recognition needs a Tesseract or RapidOCR install. With
neither, `petctl read` reports what's missing instead of raising, and the rest
of the pet carries on.
The pure parts (cleaning OCR output, formatting the reply, deciding which
engine to use given what's installed) are split out and unit tested; only
capture and the OCR call itself need a real screen.
"""
from __future__ import annotations
import re
import shutil
from typing import Callable, Optional
from .monitors import Monitor
DEFAULT_MAX_CHARS = 4000
INSTALL_HINT = (
"install one of: `pip install mss pytesseract` + `sudo apt install "
"tesseract-ocr` (fastest), or `pip install mss rapidocr-onnxruntime` "
"(no system package needed)"
)
# Lines that are almost certainly OCR noise rather than text: window chrome
# fragments, isolated punctuation, single stray characters.
_MIN_MEANINGFUL = 2
def _module_available(name: str) -> bool:
import importlib.util
try:
return importlib.util.find_spec(name) is not None
except (ImportError, ValueError):
return False
# ── pure helpers (unit tested; no screen, no OCR engine needed) ──────────────
def resolve_engine(
has_module: Callable[[str], bool] = _module_available,
which: Callable[[str], Optional[str]] = shutil.which,
) -> tuple[Optional[str], str]:
"""Pick an OCR engine from what's installed.
Returns `(engine, reason)`. *engine* is None when nothing usable is
present, and *reason* then explains what to install. Probes are injected
so this is testable on a machine with a different set of things installed.
"""
if has_module("pytesseract") and which("tesseract"):
return "pytesseract", ""
if has_module("rapidocr_onnxruntime"):
return "rapidocr", ""
if has_module("pytesseract") and not which("tesseract"):
return None, (
"pytesseract is installed but the tesseract binary isn't on PATH "
"(try: sudo apt install tesseract-ocr)"
)
return None, f"no OCR engine available — {INSTALL_HINT}"
def capture_available(has_module: Callable[[str], bool] = _module_available) -> bool:
return has_module("mss")
def clean_ocr_text(raw: str, max_chars: int = DEFAULT_MAX_CHARS) -> str:
"""Squeeze raw OCR output into something worth sending.
Screen OCR produces a lot of junk — single stray glyphs off window
borders, runs of blank lines, the same toolbar label recognised twice. All
of that costs tokens and tells the model nothing, so it goes.
"""
if not raw:
return ""
lines: list[str] = []
for line in raw.splitlines():
line = re.sub(r"[^\S\n]+", " ", line).strip()
if not line:
continue
if len(re.sub(r"[^0-9A-Za-z]", "", line)) < _MIN_MEANINGFUL:
continue
if lines and line == lines[-1]:
continue # consecutive duplicate
lines.append(line)
text = "\n".join(lines)
if max_chars and len(text) > max_chars:
text = text[: max_chars - 1].rstrip() + ""
text += "\n[truncated]"
return text
def format_reading(monitor: Optional[Monitor], text: str) -> str:
"""The tool output handed back for `petctl read`."""
where = f"monitor {monitor.number} ({monitor.name})" if monitor else "screen"
if not text.strip():
return f"[pet] read {where}: no text recognised"
return f"[pet] text on {where}:\n{text}"
# ── capture + recognition (needs a real screen) ──────────────────────────────
def capture(monitor: Monitor):
"""Grab *monitor* as a PIL image, or None if capture isn't available."""
try:
import mss
from PIL import Image
except ImportError:
return None
try:
box = {
"left": monitor.x,
"top": monitor.y,
"width": monitor.width,
"height": monitor.height,
}
with mss.mss() as sct:
shot = sct.grab(box)
return Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX")
except Exception:
return None
def _ocr(image, engine: str) -> str:
if engine == "pytesseract":
import pytesseract
# Grayscale first: tesseract is measurably better on it than on the
# colour desktop, and it's a cheap conversion.
return pytesseract.image_to_string(image.convert("L"))
if engine == "rapidocr":
import numpy as np
from rapidocr_onnxruntime import RapidOCR
result, _ = RapidOCR()(np.array(image))
if not result:
return ""
return "\n".join(line[1] for line in result)
return ""
def read_monitor(monitor: Monitor, max_chars: int = DEFAULT_MAX_CHARS) -> str:
"""OCR one screen and return the formatted tool output.
Never raises: every failure path returns a sentence explaining itself,
because the return value goes straight back to the server as the result of
a command Bolt chose to run.
"""
if not capture_available():
return f"[pet] can't capture the screen — {INSTALL_HINT}"
engine, reason = resolve_engine()
if engine is None:
return f"[pet] can't read the screen — {reason}"
image = capture(monitor)
if image is None:
return (
f"[pet] couldn't capture monitor {monitor.number} "
"(is this a Wayland session? mss needs X11)"
)
try:
raw = _ocr(image, engine)
except Exception as exc:
return f"[pet] OCR failed on monitor {monitor.number}: {exc}"
return format_reading(monitor, clean_ocr_text(raw, max_chars))
def read_monitors(monitors: list[Monitor], max_chars: int = DEFAULT_MAX_CHARS) -> str:
"""OCR several screens, splitting the character budget between them."""
if not monitors:
return "[pet] no monitor information available"
if len(monitors) == 1:
return read_monitor(monitors[0], max_chars)
share = max(400, max_chars // len(monitors))
return "\n\n".join(read_monitor(m, share) for m in monitors)
+7
View File
@@ -46,6 +46,13 @@ def run() -> int:
controller.finished.connect(thread.quit)
window.talk_requested.connect(controller.request_talk_now)
# The window owns the screen list and tells the controller about it, so
# both ends agree on what "monitor 2" means (see monitors.py).
window.monitors_changed.connect(controller.set_monitors)
window.pet_monitor_changed.connect(controller.set_pet_monitor)
# PetWindow publishes once in its constructor, which ran before those
# connections existed — so say it again now that anyone is listening.
window.publish_monitors()
window.copied.connect(lambda text: _log(f"Copied to clipboard: {text[:60]}"))
history_window = HistoryWindow(controller.history)
+188 -5
View File
@@ -20,8 +20,9 @@ from PySide6.QtGui import (
from PySide6.QtWidgets import QApplication, QWidget
from .. import config
from ..monitors import Monitor
from ..state import PetState
from .sprite import SpriteSet
from .sprite import WALK, SpriteSet
_DRAG_THRESHOLD_PX = 4
# Movement runs on its own ~30fps timer, independent of the (slower) sprite
@@ -29,6 +30,12 @@ _DRAG_THRESHOLD_PX = 4
_WANDER_TICK_MS = 33
_EMOTE_TICKS = 36 # ~1.2s per emote at the tick rate above
_NAP_OPACITY = 0.35
# How far the pet travels per walk-cycle frame. The cycle is advanced by
# distance rather than by the animation clock so a planted paw tracks backwards
# at exactly the speed the window moves forwards — drive it off a timer instead
# and the feet skate whenever PET_WANDER_SPEED doesn't happen to match the fps.
# Eight frames at 13px is a ~104px stride cycle, a bit under the pet's width.
_WALK_PIXELS_PER_FRAME = 13.0
def emote_transform(emote: str, progress: float) -> tuple[float, float, float, float]:
@@ -164,6 +171,12 @@ class SpeechBubble(QWidget):
class PetWindow(QWidget):
talk_requested = Signal()
copied = Signal(str) # bubble text the user just put on the clipboard
# The screen layout, published *to* the controller (queued, cross-thread).
# The window is the only thing allowed to ask Qt about screens, so the
# controller and the window can never disagree about what "monitor 2"
# means — see monitors.py.
monitors_changed = Signal(list) # list[monitors.Monitor]
pet_monitor_changed = Signal(int) # 0-based index the pet is standing on
def __init__(self, sprite_dir: Optional[Path] = None, size: Optional[int] = None):
super().__init__()
@@ -201,6 +214,10 @@ class PetWindow(QWidget):
self._next_wander_at = 0.0
self._bob_offset = 0
self._bob_phase = 0.0
self._walking = False
self._facing = 1 # +1 right, -1 left; the walk art is drawn facing right
self._walk_distance = 0.0
self._mirror_cache: dict[int, QPixmap] = {}
self._schedule_next_wander()
self._wander_timer = QTimer(self)
self._wander_timer.timeout.connect(self._movement_tick)
@@ -210,6 +227,18 @@ class PetWindow(QWidget):
self.set_click_through(config.PET_CLICK_THROUGH)
self._place_start_position()
self._monitors: list[Monitor] = []
self._pet_monitor: Optional[int] = None
self._last_published_pos: Optional[QPoint] = None
app = QApplication.instance()
if app is not None:
# Screens come and go — a laptop docking, a TV waking up. Republish
# rather than letting Bolt jump to a monitor that's been unplugged.
app.screenAdded.connect(lambda _s: self.publish_monitors())
app.screenRemoved.connect(lambda _s: self.publish_monitors())
app.primaryScreenChanged.connect(lambda _s: self.publish_monitors())
self.publish_monitors()
# ── placement ────────────────────────────────────────────────────────
def _place_start_position(self) -> None:
@@ -244,6 +273,8 @@ class PetWindow(QWidget):
if target is not None:
self._wander_target = target
self._commanded_move = True # overrides the idle-only rule
elif kind == "jump":
self.jump_to_monitor(int(action["monitor"]))
elif kind == "emote":
self.start_emote(action["emote"])
elif kind == "say":
@@ -378,6 +409,98 @@ class PetWindow(QWidget):
"""Stroll immediately (tray menu / anything that wants a nudge)."""
self._next_wander_at = 0.0
# ── monitors ─────────────────────────────────────────────────────────
def _build_monitors(self) -> list[Monitor]:
"""Snapshot Qt's screen list as plain dataclasses.
Full `geometry()`, not `availableGeometry()`: these coordinates are
what a screen grab gets cropped to, and a grab doesn't stop at the
taskbar. Placement uses availableGeometry separately.
"""
primary = QApplication.primaryScreen()
out = []
for index, screen in enumerate(QApplication.screens()):
geo = screen.geometry()
out.append(
Monitor(
index=index,
name=screen.name() or f"screen-{index + 1}",
x=geo.x(),
y=geo.y(),
width=geo.width(),
height=geo.height(),
primary=screen is primary,
)
)
return out
def publish_monitors(self, force: bool = True) -> None:
"""Push the current layout to whoever's listening (the controller).
*force* re-emits even when nothing changed, which is what the initial
wiring in ui/app.py needs: this window is built before the controller
exists, so the constructor's first publish goes to nobody.
"""
monitors = self._build_monitors()
changed = monitors != self._monitors
self._monitors = monitors
if changed or force:
self.monitors_changed.emit(monitors)
self._publish_pet_monitor(force=True)
def monitors(self) -> list[Monitor]:
return list(self._monitors)
def current_monitor_index(self) -> Optional[int]:
center = self.frameGeometry().center()
screens = QApplication.screens()
if not screens:
return None
screen = QApplication.screenAt(center)
if screen is not None:
try:
return screens.index(screen)
except ValueError:
pass
# Straddling a gap or dragged off the desktop entirely — fall back to
# whichever screen centre is nearest rather than reporting nothing.
best = min(
range(len(screens)),
key=lambda i: (screens[i].geometry().center() - center).manhattanLength(),
)
return best
def _publish_pet_monitor(self, force: bool = False) -> None:
index = self.current_monitor_index()
if index is None:
return
if force or index != self._pet_monitor:
self._pet_monitor = index
self.pet_monitor_changed.emit(index)
def jump_to_monitor(self, index: int) -> None:
"""Teleport to *index* (0-based, resolved by the controller) and land
with a hop. Instant rather than a stroll — Bolt asked to *jump*, and
walking between screens would take the long way across the desktop."""
screens = QApplication.screens()
if not (0 <= index < len(screens)):
return
geo = screens[index].availableGeometry()
point = self._clamp_to_screen(
QPoint(
geo.left() + (geo.width() - self.width()) // 2,
geo.top() + (geo.height() - self.height()) // 2,
),
geo,
)
self._stop_walking() # drop any stroll in flight, or it walks straight back
self.move(point)
self._schedule_next_wander()
self._publish_pet_monitor(force=True)
self.start_emote("hop")
self.update()
def _screen_geometry(self):
# screenAt() so a multi-monitor setup keeps the pet on the screen
# it's currently standing on rather than yanking it to the primary.
@@ -389,12 +512,17 @@ class PetWindow(QWidget):
self._next_wander_at = time.monotonic() + random.uniform(0.5 * base, 1.5 * base)
def _stop_walking(self) -> None:
if self._wander_target is None and not self._bob_offset:
if self._wander_target is None and not self._bob_offset and not self._walking:
return
self._wander_target = None
self._commanded_move = False
self._bob_phase = 0.0
self._bob_offset = 0
self._walking = False
self._walk_distance = 0.0
# Back to a standing frame, so the next stroll starts from a contact
# pose instead of mid-stride.
self.sprites.get(WALK).reset()
self.update()
def snap_to_edge(self) -> bool:
@@ -447,6 +575,13 @@ class PetWindow(QWidget):
wandering only happens when it's otherwise unoccupied."""
self._advance_emote()
self._wander_tick()
# Report crossing a screen boundary — by strolling, by being dragged,
# by anything. Guarded on the position actually changing so the common
# case (a stationary pet, 30x a second) costs one comparison.
position = self.pos()
if position != self._last_published_pos:
self._last_published_pos = position
self._publish_pet_monitor()
def _wander_tick(self) -> None:
# Only stroll while genuinely idle: not mid-drag, not napping, not
@@ -484,11 +619,51 @@ class PetWindow(QWidget):
self.snap_to_edge()
else:
self.move(round(here.x() + dx / distance * step), round(here.y() + dy / distance * step))
self._bob_phase += 0.45 # little walk-cycle hop
self._bob_offset = int(round(-2.5 * abs(math.sin(self._bob_phase))))
self._advance_walk(dx, dy, step)
self.update()
self._reposition_bubble()
def _advance_walk(self, dx: float, dy: float, step: float) -> None:
"""Drive the walk cycle from distance travelled (see the constant).
Falls back to the old bob-in-code if there's no walk art, so a sprite
folder without a walk/ directory still looks like it's moving rather
than sliding perfectly flat.
"""
self._walking = True
# Only turn on meaningful horizontal travel: a near-vertical stroll
# would otherwise flip him back and forth on rounding noise.
if abs(dx) > 1.0:
self._facing = 1 if dx > 0 else -1
if not self.sprites.has(WALK):
self._bob_phase += 0.45
self._bob_offset = int(round(-2.5 * abs(math.sin(self._bob_phase))))
return
self._bob_offset = 0 # the walk frames carry their own weight shift
self._walk_distance += step
while self._walk_distance >= _WALK_PIXELS_PER_FRAME:
self._walk_distance -= _WALK_PIXELS_PER_FRAME
self.sprites.get(WALK).advance()
def _animation_key(self):
"""Walking overrides the state animation — but only while genuinely
idle-and-moving, so he doesn't trot on the spot mid-sentence."""
if self._walking and self.sprites.has(WALK):
return WALK
return self._current_state
def _oriented(self, pixmap: Optional[QPixmap]) -> Optional[QPixmap]:
"""Mirror the (right-facing) walk art when he's heading left. Cached
per source frame — flipping on every paint would be wasteful at 30fps."""
if pixmap is None or self._facing >= 0:
return pixmap
key = pixmap.cacheKey()
mirrored = self._mirror_cache.get(key)
if mirrored is None:
mirrored = pixmap.transformed(QTransform().scale(-1, 1), Qt.SmoothTransformation)
self._mirror_cache[key] = mirrored
return mirrored
# ── state / speech ──────────────────────────────────────────────────
def set_state(self, state: PetState) -> None:
@@ -514,6 +689,11 @@ class PetWindow(QWidget):
# ── animation ────────────────────────────────────────────────────────
def _advance_frame(self) -> None:
# While walking the cycle is stepped by _advance_walk from distance
# travelled; letting this timer also advance it would double-step it
# and put the feet out of sync with the movement.
if self._walking and self.sprites.has(WALK):
return
self.sprites.get(self._current_state).advance()
self.update()
@@ -521,7 +701,10 @@ class PetWindow(QWidget):
painter = QPainter(self)
painter.setRenderHint(QPainter.Antialiasing)
painter.setRenderHint(QPainter.SmoothPixmapTransform)
pixmap: Optional[QPixmap] = self.sprites.get(self._current_state).current()
key = self._animation_key()
pixmap: Optional[QPixmap] = self.sprites.get(key).current()
if key == WALK:
pixmap = self._oriented(pixmap)
if pixmap is None:
self._apply_input_mask(None, 0, 0)
return
+35 -6
View File
@@ -95,17 +95,46 @@ def _load_frames_from_dir(directory: Path, size: int) -> list[QPixmap]:
return frames
WALK = "walk"
# Animations that aren't pipeline states. Walking is a property of *movement*,
# orthogonal to whether the pet is idle/listening/talking, so it deliberately
# isn't a PetState — state.py stays a description of the conversation, not of
# the body. Loaded the same way, keyed by name.
EXTRA_ANIMATIONS = (WALK,)
class SpriteSet:
"""All animations for every PetState, loaded from *sprite_dir*."""
"""All animations for every PetState, plus the extras, from *sprite_dir*."""
def __init__(self, sprite_dir: Path = DEFAULT_SPRITE_DIR, size: int = 160):
self.size = size
self._animations: dict[PetState, SpriteAnimation] = {}
self._animations: dict[str, SpriteAnimation] = {}
self._loaded: set[str] = set() # keys backed by real art, not placeholders
for state in PetState:
frames = _load_frames_from_dir(sprite_dir / state.value, size)
if not frames:
if frames:
self._loaded.add(state.value)
else:
frames = _placeholder_frames(state, size)
self._animations[state] = SpriteAnimation(frames)
self._animations[state.value] = SpriteAnimation(frames)
for name in EXTRA_ANIMATIONS:
frames = _load_frames_from_dir(sprite_dir / name, size)
if frames:
self._loaded.add(name)
self._animations[name] = SpriteAnimation(frames)
def get(self, state: PetState) -> SpriteAnimation:
return self._animations[state]
@staticmethod
def _key(key) -> str:
return key.value if isinstance(key, PetState) else str(key)
def get(self, key) -> SpriteAnimation:
"""Animation for a PetState or an extra name. Unknown/absent extras
fall back to idle, so a sprite folder with no walk/ still runs."""
return self._animations.get(self._key(key)) or self._animations[PetState.IDLE.value]
def has(self, key) -> bool:
"""True only when real frames were found — the caller uses this to
decide whether to use an extra animation at all, rather than being
handed a placeholder blob that looks nothing like walking."""
return self._key(key) in self._loaded