themajesticmagician
3a0959f55d
Latency: replies are spoken sentence-by-sentence off the desk API's NDJSON endpoint, so the wait is time-to-first-sentence rather than the whole model call, and Deepgram's live websocket transcribes while you're still talking instead of uploading the WAV afterwards. Both fall back invisibly — a stream that fails before anything was said drops to converse(), and a socket that never opens just means the old one-shot path. Speaking lived in four near-copies in the controller (a reply, a holding line, a streamed sentence, a dialogue scene) that had already drifted: one didn't arm barge-in, another skipped the follow-up rule. It's now speech.Speaker plus an Utterance describing the policy differences, with collaborators injected so the whole of it tests without Qt or audio. The mouth follows the audio rather than a timer: tts.level_of reduces each PCM frame to a 0..1 loudness on a sqrt curve (speech sits well below peak, and a linear map leaves the mouth barely open during normal talking) and that indexes the talking frames, which the sprite script now draws as an openness ramp. Offline pyttsx3 has no waveform, so stale levels hand control back to the timed loop instead of freezing the mouth mid-syllable. Also: the pet starts where you left it (ignoring positions on monitors that are no longer connected, since restoring those faithfully is how it ends up somewhere unreachable), and `python -m bolt_pet --doctor` is a preflight that says what to do about each problem rather than only what's wrong. tests/test_pipeline_smoke.py breaks the pure-logic rule on purpose. Every unit test passed all week while notifications sat unspoken for minutes, the pet said things twice and [laughing] got read aloud — each an interaction between two individually-correct units. It drives whole turns against a real HTTP server on a loopback port, faking only the mic and the speakers. It found a NameError in the paint path that would have fired on every repaint while talking. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
823 lines
35 KiB
Python
823 lines
35 KiB
Python
"""The pet itself: a frameless, translucent, always-on-top window that
|
|
renders the current sprite animation, wanders the desktop on its own while
|
|
idle, walks/emotes on command from the server (see pet_actions.py), can be
|
|
dragged around, dims when napping, and turns a plain (non-drag) click into a
|
|
"talk now" request.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import random
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from PySide6.QtCore import QPoint, Qt, QTimer, Signal
|
|
from PySide6.QtGui import (
|
|
QColor, QCursor, QFont, QFontMetrics, QPainter, QPainterPath, QPixmap, QRegion, QTransform,
|
|
)
|
|
from PySide6.QtWidgets import QApplication, QWidget
|
|
|
|
from .. import config, window_state
|
|
from ..monitors import Monitor
|
|
from ..state import PetState
|
|
from .sprite import WALK, SpriteSet
|
|
|
|
_DRAG_THRESHOLD_PX = 4
|
|
# Movement runs on its own ~30fps timer, independent of the (slower) sprite
|
|
# animation timer, so a stroll looks smooth even at IDLE_ANIMATION_FPS=6.
|
|
_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
|
|
# How long a loudness level stays believable. The audio thread sends one per
|
|
# ~30ms while a clip plays; if they stop arriving (offline TTS has no envelope,
|
|
# or playback died) the mouth must not stay frozen mid-syllable, so after this
|
|
# long the ordinary looping animation takes back over.
|
|
_MOUTH_STALE_SECONDS = 0.35
|
|
|
|
|
|
def emote_transform(emote: str, progress: float) -> tuple[float, float, float, float]:
|
|
"""(dx, dy, rotation_degrees, scale) for an emote at *progress* 0..1.
|
|
|
|
Pure maths, deliberately separate from paintEvent so the motion curves can
|
|
be unit tested (and so adding an emote doesn't mean touching painting
|
|
code). Every emote must return to (0, 0, 0, 1) at progress 1.0, otherwise
|
|
the pet ends up permanently askew.
|
|
"""
|
|
progress = min(max(progress, 0.0), 1.0)
|
|
fade = math.sin(math.pi * progress) # 0 -> 1 -> 0, so it always lands home
|
|
tau = 2 * math.pi
|
|
if emote == "wave":
|
|
return 0.0, 0.0, 14.0 * fade * math.sin(tau * 2 * progress), 1.0
|
|
if emote in ("hop", "bounce"):
|
|
hops = 2 if emote == "hop" else 3
|
|
return 0.0, -22.0 * fade * abs(math.sin(math.pi * hops * progress)), 0.0, 1.0
|
|
if emote == "spin":
|
|
return 0.0, 0.0, 360.0 * progress % 360.0, 1.0
|
|
if emote == "nod":
|
|
return 0.0, 10.0 * fade * math.sin(tau * 2 * progress), 0.0, 1.0 - 0.05 * fade
|
|
if emote in ("shake", "wiggle"):
|
|
return 14.0 * fade * math.sin(tau * 3 * progress), 0.0, 0.0, 1.0
|
|
return 0.0, 0.0, 0.0, 1.0
|
|
|
|
|
|
class SpeechBubble(QWidget):
|
|
"""Small translucent word-bubble shown above the pet while it talks."""
|
|
|
|
_MAX_WIDTH = 260
|
|
_PADDING = 10
|
|
|
|
copied = Signal(str)
|
|
|
|
def __init__(self, parent: Optional[QWidget] = None):
|
|
super().__init__(parent, Qt.FramelessWindowHint | Qt.Tool)
|
|
self.setAttribute(Qt.WA_TranslucentBackground)
|
|
self.setAttribute(Qt.WA_ShowWithoutActivating)
|
|
self._text = ""
|
|
self._font = QFont()
|
|
self._font.setPointSize(10)
|
|
self._flash = "" # transient overlay ("Copied") drawn over the text
|
|
self._hide_timer = QTimer(self)
|
|
self._hide_timer.setSingleShot(True)
|
|
self._hide_timer.timeout.connect(self.hide)
|
|
self._flash_timer = QTimer(self)
|
|
self._flash_timer.setSingleShot(True)
|
|
self._flash_timer.timeout.connect(self._clear_flash)
|
|
self.setToolTip("Click to copy")
|
|
self.setCursor(Qt.PointingHandCursor)
|
|
self.hide()
|
|
|
|
@property
|
|
def text(self) -> str:
|
|
return self._text
|
|
|
|
def show_text(self, text: str, duration_ms: int = 6000) -> None:
|
|
text = (text or "").strip()
|
|
if not text:
|
|
self.hide()
|
|
return
|
|
self._text = text
|
|
self._relayout()
|
|
self.show()
|
|
self.raise_()
|
|
self._hide_timer.start(duration_ms)
|
|
|
|
# ── click to copy ────────────────────────────────────────────────────
|
|
# The bubble hides itself after a few seconds, which is fine for chat and
|
|
# awful for anything you needed to keep (a path, a number, a command's
|
|
# output). One click puts it on the clipboard; the tray's History window
|
|
# has the rest.
|
|
|
|
def mousePressEvent(self, event) -> None:
|
|
if event.button() != Qt.LeftButton or not self._text:
|
|
return
|
|
QApplication.clipboard().setText(self._text)
|
|
self.copied.emit(self._text)
|
|
self._flash = "Copied to clipboard"
|
|
self.update()
|
|
self._flash_timer.start(900)
|
|
self._hide_timer.start(2500) # linger a moment so the flash is visible
|
|
|
|
def _clear_flash(self) -> None:
|
|
self._flash = ""
|
|
self.update()
|
|
|
|
def _wrapped_lines(self) -> list[str]:
|
|
metrics = QFontMetrics(self._font)
|
|
words = self._text.split()
|
|
lines: list[str] = []
|
|
current = ""
|
|
max_text_width = self._MAX_WIDTH - 2 * self._PADDING
|
|
for word in words:
|
|
candidate = f"{current} {word}".strip()
|
|
if metrics.horizontalAdvance(candidate) <= max_text_width or not current:
|
|
current = candidate
|
|
else:
|
|
lines.append(current)
|
|
current = word
|
|
if current:
|
|
lines.append(current)
|
|
return lines[:6] # don't let a huge reply turn into a wall of bubble
|
|
|
|
def _relayout(self) -> None:
|
|
metrics = QFontMetrics(self._font)
|
|
lines = self._wrapped_lines()
|
|
text_width = max((metrics.horizontalAdvance(line) for line in lines), default=0)
|
|
width = min(self._MAX_WIDTH, text_width + 2 * self._PADDING)
|
|
height = metrics.height() * len(lines) + 2 * self._PADDING
|
|
self.resize(width, height)
|
|
|
|
def paintEvent(self, _event) -> None:
|
|
painter = QPainter(self)
|
|
painter.setRenderHint(QPainter.Antialiasing)
|
|
path = QPainterPath()
|
|
path.addRoundedRect(0, 0, self.width(), self.height(), 10, 10)
|
|
painter.fillPath(path, QColor(30, 30, 35, 230))
|
|
painter.setFont(self._font)
|
|
metrics = QFontMetrics(self._font)
|
|
if self._flash:
|
|
painter.setPen(QColor(150, 230, 170))
|
|
painter.drawText(self.rect(), Qt.AlignCenter, self._flash)
|
|
return
|
|
painter.setPen(QColor(240, 240, 245))
|
|
y = self._PADDING + metrics.ascent()
|
|
for line in self._wrapped_lines():
|
|
painter.drawText(self._PADDING, y, line)
|
|
y += metrics.height()
|
|
|
|
|
|
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__()
|
|
flags = Qt.FramelessWindowHint | Qt.Tool
|
|
if config.PET_ALWAYS_ON_TOP:
|
|
flags |= Qt.WindowStaysOnTopHint
|
|
self.setWindowFlags(flags)
|
|
self.setAttribute(Qt.WA_TranslucentBackground)
|
|
|
|
self.sprites = SpriteSet(sprite_dir or (Path(__file__).resolve().parent.parent / "assets" / "sprites"),
|
|
size or config.PET_SIZE)
|
|
self.resize(self.sprites.size, self.sprites.size)
|
|
self._current_state = PetState.IDLE
|
|
|
|
self._drag_offset: Optional[QPoint] = None
|
|
self._press_pos: Optional[QPoint] = None
|
|
self._dragged = False
|
|
|
|
self._bubble = SpeechBubble()
|
|
self._bubble.copied.connect(self.copied)
|
|
|
|
self._napping = False
|
|
self._emote: Optional[str] = None
|
|
self._emote_tick = 0
|
|
self._mask_key = None
|
|
|
|
self._anim_timer = QTimer(self)
|
|
self._anim_timer.timeout.connect(self._advance_frame)
|
|
fps = max(1.0, config.IDLE_ANIMATION_FPS)
|
|
self._anim_timer.start(int(1000 / fps))
|
|
|
|
self._wander_enabled = config.PET_WANDER
|
|
self._wander_target: Optional[QPoint] = None
|
|
self._commanded_move = False # a petctl move — happens even mid-conversation
|
|
self._next_wander_at = 0.0
|
|
self._bob_offset = 0
|
|
# Lip-sync: loudness of what is playing right now, and when it arrived.
|
|
self._mouth_level: Optional[float] = None
|
|
self._mouth_at = 0.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)
|
|
self._wander_timer.start(_WANDER_TICK_MS)
|
|
|
|
self._click_through = False
|
|
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:
|
|
screen = QApplication.primaryScreen()
|
|
geo = screen.availableGeometry() if screen else None
|
|
try:
|
|
x = int(config.PET_START_X) if config.PET_START_X else None
|
|
y = int(config.PET_START_Y) if config.PET_START_Y else None
|
|
except ValueError:
|
|
x = y = None
|
|
if x is None and y is None and config.PET_REMEMBER_POSITION:
|
|
remembered = window_state.load()
|
|
# Only if it still lands on a screen that exists — the usual reason
|
|
# a saved position is stale is that the monitor it was on has been
|
|
# unplugged, and restoring it faithfully would hide the pet.
|
|
if remembered is not None:
|
|
rectangles = [
|
|
(s.availableGeometry().left(), s.availableGeometry().top(),
|
|
s.availableGeometry().right(), s.availableGeometry().bottom())
|
|
for s in QApplication.screens()
|
|
]
|
|
if window_state.is_visible_on(*remembered, self.width(), rectangles):
|
|
x, y = remembered
|
|
if geo is not None:
|
|
x = geo.right() - self.width() - 40 if x is None else x
|
|
y = geo.bottom() - self.height() - 60 if y is None else y
|
|
self.move(x or 0, y or 0)
|
|
self._reposition_bubble()
|
|
|
|
def _reposition_bubble(self) -> None:
|
|
top_left = self.geometry().topLeft()
|
|
self._bubble.move(
|
|
top_left.x() + self.width() // 2 - self._bubble.width() // 2,
|
|
top_left.y() - self._bubble.height() - 8,
|
|
)
|
|
|
|
# ── server-driven actions (petctl) ───────────────────────────────────
|
|
|
|
def apply_action(self, action: dict) -> None:
|
|
"""Perform one parsed petctl action (see pet_actions.py). Called on
|
|
the UI thread via a queued signal from the controller."""
|
|
kind = action.get("action")
|
|
if kind == "move":
|
|
target = self._resolve_move_target(action)
|
|
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":
|
|
self.say(action["text"])
|
|
elif kind == "wander":
|
|
self.set_wander_enabled(bool(action["enabled"]))
|
|
elif kind == "nap":
|
|
self.set_napping(bool(action["enabled"]))
|
|
|
|
def _resolve_move_target(self, action: dict) -> Optional[QPoint]:
|
|
geo = self._screen_geometry()
|
|
if "x" in action and "y" in action:
|
|
point = QPoint(int(action["x"]), int(action["y"]))
|
|
return self._clamp_to_screen(point, geo)
|
|
anchor = action.get("anchor")
|
|
if anchor == "cursor":
|
|
cursor = QCursor.pos()
|
|
return self._clamp_to_screen(
|
|
QPoint(cursor.x() - self.width() // 2, cursor.y() - self.height() // 2), geo
|
|
)
|
|
if geo is None:
|
|
return None
|
|
if anchor == "random":
|
|
return self._pick_wander_target()
|
|
margin = config.PET_WANDER_MARGIN
|
|
left, right = geo.left() + margin, geo.right() - self.width() - margin
|
|
top, bottom = geo.top() + margin, geo.bottom() - self.height() - margin
|
|
middle_x = geo.left() + (geo.width() - self.width()) // 2
|
|
middle_y = geo.top() + (geo.height() - self.height()) // 2
|
|
positions = {
|
|
"top-left": (left, top), "top": (middle_x, top), "top-right": (right, top),
|
|
"left": (left, middle_y), "center": (middle_x, middle_y), "right": (right, middle_y),
|
|
"bottom-left": (left, bottom), "bottom": (middle_x, bottom), "bottom-right": (right, bottom),
|
|
}
|
|
if anchor not in positions:
|
|
return None
|
|
return QPoint(*positions[anchor])
|
|
|
|
def _clamp_to_screen(self, point: QPoint, geo) -> QPoint:
|
|
if geo is None:
|
|
return point
|
|
x = min(max(point.x(), geo.left()), max(geo.left(), geo.right() - self.width()))
|
|
y = min(max(point.y(), geo.top()), max(geo.top(), geo.bottom() - self.height()))
|
|
return QPoint(x, y)
|
|
|
|
# ── emotes ───────────────────────────────────────────────────────────
|
|
|
|
def start_emote(self, emote: str) -> None:
|
|
self._emote = emote
|
|
self._emote_tick = 0
|
|
self.update()
|
|
|
|
def _advance_emote(self) -> None:
|
|
if self._emote is None:
|
|
return
|
|
self._emote_tick += 1
|
|
if self._emote_tick > _EMOTE_TICKS:
|
|
self._emote = None
|
|
self._emote_tick = 0
|
|
self.update()
|
|
|
|
def _emote_transform(self) -> tuple[float, float, float, float]:
|
|
if self._emote is None:
|
|
return 0.0, 0.0, 0.0, 1.0
|
|
return emote_transform(self._emote, self._emote_tick / _EMOTE_TICKS)
|
|
|
|
# ── napping (quiet hours / do-not-disturb) ───────────────────────────
|
|
|
|
def set_napping(self, napping: bool) -> None:
|
|
"""Dim and stand still. Purely cosmetic here — the controller is what
|
|
actually suppresses proactive speech."""
|
|
if napping == self._napping:
|
|
return
|
|
self._napping = napping
|
|
self.setWindowOpacity(_NAP_OPACITY if napping else 1.0)
|
|
if napping:
|
|
self._stop_walking()
|
|
self.update()
|
|
|
|
@property
|
|
def napping(self) -> bool:
|
|
return self._napping
|
|
|
|
# ── mouse transparency ───────────────────────────────────────────────
|
|
|
|
def set_click_through(self, enabled: bool) -> None:
|
|
"""When on, the pet ignores the mouse entirely (tray-only control) —
|
|
for when it's parked over something you need to click a lot."""
|
|
self._click_through = enabled
|
|
self.setAttribute(Qt.WA_TransparentForMouseEvents, enabled)
|
|
if enabled:
|
|
self.clearMask()
|
|
self._mask_key = None
|
|
else:
|
|
self._mask_key = None # force the shaped mask to be rebuilt
|
|
|
|
@property
|
|
def click_through(self) -> bool:
|
|
return self._click_through
|
|
|
|
def _apply_input_mask(self, pixmap: Optional[QPixmap], x: int, y: int) -> None:
|
|
"""Restrict the window to the sprite's opaque pixels, so the square
|
|
window's transparent corners stop swallowing clicks meant for what's
|
|
underneath. Rebuilt only when the frame actually changes — the mask
|
|
is derived from the pixmap's alpha, which isn't free."""
|
|
if self._click_through or not config.PET_SHAPED_INPUT:
|
|
return
|
|
if pixmap is None:
|
|
if self._mask_key is not None:
|
|
self.clearMask()
|
|
self._mask_key = None
|
|
return
|
|
key = (pixmap.cacheKey(), x, y)
|
|
if key == self._mask_key:
|
|
return
|
|
self._mask_key = key
|
|
try:
|
|
region = QRegion(pixmap.mask())
|
|
region.translate(x, y)
|
|
self.setMask(region)
|
|
except Exception:
|
|
self.clearMask() # a sprite without an alpha channel — never mind
|
|
|
|
# ── wandering ────────────────────────────────────────────────────────
|
|
|
|
def set_wander_enabled(self, enabled: bool) -> None:
|
|
self._wander_enabled = enabled
|
|
if not enabled:
|
|
self._stop_walking()
|
|
|
|
def wander_now(self) -> None:
|
|
"""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.
|
|
screen = QApplication.screenAt(self.frameGeometry().center()) or QApplication.primaryScreen()
|
|
return screen.availableGeometry() if screen else None
|
|
|
|
def _schedule_next_wander(self) -> None:
|
|
base = max(1.0, config.PET_WANDER_INTERVAL_SECONDS)
|
|
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 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:
|
|
"""If the pet has come to rest near a screen edge, tuck it flush
|
|
against it — a desktop pet parked 11px off the taskbar looks like a
|
|
bug. Returns True if it moved."""
|
|
geo = self._screen_geometry()
|
|
if geo is None or not config.PET_EDGE_SNAP:
|
|
return False
|
|
margin = config.PET_SNAP_MARGIN
|
|
here = self.pos()
|
|
x, y = here.x(), here.y()
|
|
if abs(x - geo.left()) <= margin:
|
|
x = geo.left()
|
|
elif abs(geo.right() - (x + self.width())) <= margin:
|
|
x = geo.right() - self.width() + 1
|
|
if abs(y - geo.top()) <= margin:
|
|
y = geo.top()
|
|
elif abs(geo.bottom() - (y + self.height())) <= margin:
|
|
y = geo.bottom() - self.height() + 1
|
|
if (x, y) == (here.x(), here.y()):
|
|
return False
|
|
self.move(x, y)
|
|
self._reposition_bubble()
|
|
return True
|
|
|
|
def _pick_wander_target(self) -> Optional[QPoint]:
|
|
geo = self._screen_geometry()
|
|
if geo is None:
|
|
return None
|
|
margin = config.PET_WANDER_MARGIN
|
|
min_x, max_x = geo.left() + margin, geo.right() - self.width() - margin
|
|
min_y, max_y = geo.top() + margin, geo.bottom() - self.height() - margin
|
|
if max_x <= min_x or max_y <= min_y: # pet bigger than the screen
|
|
return None
|
|
here = self.pos()
|
|
target = QPoint(random.randint(min_x, max_x), random.randint(min_y, max_y))
|
|
dx, dy = target.x() - here.x(), target.y() - here.y()
|
|
distance = math.hypot(dx, dy)
|
|
limit = max(1.0, config.PET_WANDER_MAX_DISTANCE)
|
|
if distance > limit: # shorten the trip rather than sprinting the diagonal
|
|
scale = limit / distance
|
|
target = QPoint(round(here.x() + dx * scale), round(here.y() + dy * scale))
|
|
elif distance < 8: # already there — not worth a stroll
|
|
return None
|
|
return target
|
|
|
|
def _movement_tick(self) -> None:
|
|
"""One timer, two jobs — emotes play whatever the pet is doing, while
|
|
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
|
|
# talking/listening, and not while a speech bubble is up (it would walk
|
|
# out from under it). A commanded `petctl move` ignores all of that
|
|
# except the drag — if Bolt says go, it goes.
|
|
busy = (
|
|
not self._wander_enabled
|
|
or self._napping
|
|
or self._current_state != PetState.IDLE
|
|
or self._bubble.isVisible()
|
|
)
|
|
if self._drag_offset is not None or (busy and not self._commanded_move):
|
|
self._stop_walking()
|
|
self._schedule_next_wander() # settle first, then wander
|
|
return
|
|
|
|
if self._wander_target is None:
|
|
if time.monotonic() < self._next_wander_at:
|
|
return
|
|
self._wander_target = self._pick_wander_target()
|
|
if self._wander_target is None:
|
|
self._schedule_next_wander()
|
|
return
|
|
|
|
here = self.pos()
|
|
dx = self._wander_target.x() - here.x()
|
|
dy = self._wander_target.y() - here.y()
|
|
distance = math.hypot(dx, dy)
|
|
step = max(1.0, config.PET_WANDER_SPEED * _WANDER_TICK_MS / 1000.0)
|
|
if distance <= step:
|
|
self.move(self._wander_target)
|
|
self._stop_walking()
|
|
self._schedule_next_wander()
|
|
self.snap_to_edge()
|
|
else:
|
|
self.move(round(here.x() + dx / distance * step), round(here.y() + dy / distance * step))
|
|
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:
|
|
if state == self._current_state:
|
|
return
|
|
self._current_state = state
|
|
self.sprites.get(state).reset()
|
|
if state != PetState.IDLE and not self._commanded_move:
|
|
# Stand still while listening/thinking/talking — but not if Bolt
|
|
# just told it to walk somewhere: that command arrives mid-turn,
|
|
# and the reply (-> TALKING) lands a moment later.
|
|
self._stop_walking()
|
|
self.update()
|
|
|
|
def say(self, text: str, duration_ms: int = 6000) -> None:
|
|
self._reposition_bubble()
|
|
self._bubble.show_text(text, duration_ms)
|
|
|
|
def closeEvent(self, event) -> None:
|
|
self._bubble.close()
|
|
super().closeEvent(event)
|
|
|
|
# ── animation ────────────────────────────────────────────────────────
|
|
|
|
def set_mouth(self, level: float) -> None:
|
|
"""How loud the pet is *right now* (0..1), straight off the PCM going
|
|
to the speakers (audio/tts.level_of).
|
|
|
|
The talking frames are ordered by mouth openness, so this indexes them
|
|
directly: the mouth moves with the actual waveform instead of flapping
|
|
on a timer, which is the difference between a talking sprite and a
|
|
dubbed one."""
|
|
self._mouth_level = max(0.0, min(1.0, float(level)))
|
|
self._mouth_at = time.monotonic()
|
|
if self._current_state == PetState.TALKING:
|
|
self.update()
|
|
|
|
def _mouth_frame(self, animation) -> Optional[QPixmap]:
|
|
"""The frame matching the current loudness, or None to use the timer.
|
|
|
|
Falls back the moment the levels go stale — offline TTS has no
|
|
envelope, and a mouth frozen mid-syllable is worse than a timed loop."""
|
|
if self._mouth_level is None or animation is None:
|
|
return None
|
|
if time.monotonic() - self._mouth_at > _MOUTH_STALE_SECONDS:
|
|
return None
|
|
frames = animation.frames
|
|
if len(frames) < 2:
|
|
return None
|
|
index = int(round(self._mouth_level * (len(frames) - 1)))
|
|
return frames[max(0, min(len(frames) - 1, index))]
|
|
|
|
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()
|
|
|
|
def paintEvent(self, _event) -> None:
|
|
painter = QPainter(self)
|
|
painter.setRenderHint(QPainter.Antialiasing)
|
|
painter.setRenderHint(QPainter.SmoothPixmapTransform)
|
|
key = self._animation_key()
|
|
animation = self.sprites.get(key)
|
|
pixmap: Optional[QPixmap] = None
|
|
if key == PetState.TALKING:
|
|
pixmap = self._mouth_frame(animation)
|
|
if pixmap is None:
|
|
pixmap = animation.current()
|
|
if key == WALK:
|
|
pixmap = self._oriented(pixmap)
|
|
if pixmap is None:
|
|
self._apply_input_mask(None, 0, 0)
|
|
return
|
|
# Non-square source art (e.g. the Kenney robot sprites) keeps its
|
|
# aspect ratio when scaled in SpriteSet, so it may be narrower or
|
|
# shorter than the (square) window — center it either way.
|
|
x = (self.width() - pixmap.width()) // 2
|
|
y = (self.height() - pixmap.height()) // 2 + self._bob_offset
|
|
# The input mask tracks the resting position, not the emote/bob
|
|
# offset: rebuilding it every frame of a spin would be both expensive
|
|
# and visibly janky, and the offsets are only a few pixels.
|
|
self._apply_input_mask(pixmap, x, (self.height() - pixmap.height()) // 2)
|
|
|
|
dx, dy, angle, scale = self._emote_transform()
|
|
if (dx, dy, angle, scale) == (0.0, 0.0, 0.0, 1.0):
|
|
painter.drawPixmap(x, y, pixmap)
|
|
return
|
|
# Rotate/scale about the sprite's own center so a spin doesn't orbit
|
|
# the window's corner.
|
|
center_x = x + pixmap.width() / 2
|
|
center_y = y + pixmap.height() / 2
|
|
transform = QTransform()
|
|
transform.translate(center_x + dx, center_y + dy)
|
|
transform.rotate(angle)
|
|
transform.scale(scale, scale)
|
|
transform.translate(-pixmap.width() / 2, -pixmap.height() / 2)
|
|
painter.setTransform(transform)
|
|
painter.drawPixmap(0, 0, pixmap)
|
|
|
|
# ── drag / click-to-talk ─────────────────────────────────────────────
|
|
|
|
def mousePressEvent(self, event) -> None:
|
|
if event.button() == Qt.LeftButton:
|
|
global_pos = event.globalPosition().toPoint()
|
|
self._drag_offset = global_pos - self.frameGeometry().topLeft()
|
|
self._press_pos = global_pos
|
|
self._dragged = False
|
|
self._stop_walking() # grabbing it interrupts a stroll at once
|
|
|
|
def mouseMoveEvent(self, event) -> None:
|
|
if self._drag_offset is None:
|
|
return
|
|
global_pos = event.globalPosition().toPoint()
|
|
self.move(global_pos - self._drag_offset)
|
|
self._reposition_bubble()
|
|
if (global_pos - self._press_pos).manhattanLength() > _DRAG_THRESHOLD_PX:
|
|
self._dragged = True
|
|
|
|
def mouseReleaseEvent(self, event) -> None:
|
|
if event.button() != Qt.LeftButton:
|
|
return
|
|
was_click = not self._dragged
|
|
self._drag_offset = None
|
|
self._press_pos = None
|
|
if not was_click and config.PET_REMEMBER_POSITION:
|
|
position = self.geometry().topLeft()
|
|
window_state.save(position.x(), position.y())
|
|
if was_click:
|
|
self.talk_requested.emit()
|
|
else:
|
|
self.snap_to_edge() # dropped near an edge -> tuck it flush
|