forked from Rockachopa/Timmy-time-dashboard
Kimi Agent
996ccec170
Co-authored-by: Kimi Agent <kimi@timmy.local> Co-committed-by: Kimi Agent <kimi@timmy.local>
264 lines
8.6 KiB
Python
264 lines
8.6 KiB
Python
"""Pip the Familiar — a creature with its own small mind.
|
|
|
|
Pip is a glowing sprite who lives in the Workshop independently of Timmy.
|
|
He has a behavioral state machine that makes the room feel alive:
|
|
|
|
SLEEPING → WAKING → WANDERING → INVESTIGATING → BORED → SLEEPING
|
|
|
|
Special states triggered by Timmy's cognitive signals:
|
|
ALERT — confidence drops below 0.3
|
|
PLAYFUL — Timmy is amused / energized
|
|
HIDING — unknown visitor + Timmy uncertain
|
|
|
|
The backend tracks Pip's *logical* state; the browser handles movement
|
|
interpolation and particle rendering.
|
|
"""
|
|
|
|
import logging
|
|
import random
|
|
import time
|
|
from dataclasses import asdict, dataclass, field
|
|
from enum import StrEnum
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# States
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class PipState(StrEnum):
|
|
"""Pip's behavioral states."""
|
|
|
|
SLEEPING = "sleeping"
|
|
WAKING = "waking"
|
|
WANDERING = "wandering"
|
|
INVESTIGATING = "investigating"
|
|
BORED = "bored"
|
|
# Special states
|
|
ALERT = "alert"
|
|
PLAYFUL = "playful"
|
|
HIDING = "hiding"
|
|
|
|
|
|
# States from which Pip can be interrupted by special triggers
|
|
_INTERRUPTIBLE = frozenset(
|
|
{
|
|
PipState.SLEEPING,
|
|
PipState.WANDERING,
|
|
PipState.BORED,
|
|
PipState.WAKING,
|
|
}
|
|
)
|
|
|
|
# How long each state lasts before auto-transitioning (seconds)
|
|
_STATE_DURATIONS: dict[PipState, tuple[float, float]] = {
|
|
PipState.SLEEPING: (120.0, 300.0), # 2-5 min
|
|
PipState.WAKING: (1.5, 2.5),
|
|
PipState.WANDERING: (15.0, 45.0),
|
|
PipState.INVESTIGATING: (8.0, 12.0),
|
|
PipState.BORED: (20.0, 40.0),
|
|
PipState.ALERT: (10.0, 20.0),
|
|
PipState.PLAYFUL: (8.0, 15.0),
|
|
PipState.HIDING: (15.0, 30.0),
|
|
}
|
|
|
|
# Default position near the fireplace
|
|
_FIREPLACE_POS = (2.1, 0.5, -1.3)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schema
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class PipSnapshot:
|
|
"""Serialisable snapshot of Pip's current state."""
|
|
|
|
name: str = "Pip"
|
|
state: str = "sleeping"
|
|
position: tuple[float, float, float] = _FIREPLACE_POS
|
|
mood_mirror: str = "calm"
|
|
since: float = field(default_factory=time.monotonic)
|
|
|
|
def to_dict(self) -> dict:
|
|
"""Public dict for API / WebSocket / state file consumers."""
|
|
d = asdict(self)
|
|
d["position"] = list(d["position"])
|
|
# Convert monotonic timestamp to duration
|
|
d["state_duration_s"] = round(time.monotonic() - d.pop("since"), 1)
|
|
return d
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Familiar
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class Familiar:
|
|
"""Pip's behavioral AI — a tiny state machine driven by events and time.
|
|
|
|
Usage::
|
|
|
|
pip_familiar.on_event("visitor_entered")
|
|
pip_familiar.on_mood_change("energized")
|
|
state = pip_familiar.tick() # call periodically
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._state = PipState.SLEEPING
|
|
self._entered_at = time.monotonic()
|
|
self._duration = random.uniform(*_STATE_DURATIONS[PipState.SLEEPING])
|
|
self._mood_mirror = "calm"
|
|
self._pending_mood: str | None = None
|
|
self._mood_change_at: float = 0.0
|
|
self._position = _FIREPLACE_POS
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public API
|
|
# ------------------------------------------------------------------
|
|
|
|
@property
|
|
def state(self) -> PipState:
|
|
return self._state
|
|
|
|
@property
|
|
def mood_mirror(self) -> str:
|
|
return self._mood_mirror
|
|
|
|
def snapshot(self) -> PipSnapshot:
|
|
"""Current state as a serialisable snapshot."""
|
|
return PipSnapshot(
|
|
state=self._state.value,
|
|
position=self._position,
|
|
mood_mirror=self._mood_mirror,
|
|
since=self._entered_at,
|
|
)
|
|
|
|
def tick(self, now: float | None = None) -> PipState:
|
|
"""Advance the state machine. Call periodically (e.g. every second).
|
|
|
|
Returns the (possibly new) state.
|
|
"""
|
|
now = now if now is not None else time.monotonic()
|
|
|
|
# Apply delayed mood mirror (3-second lag)
|
|
if self._pending_mood and now >= self._mood_change_at:
|
|
self._mood_mirror = self._pending_mood
|
|
self._pending_mood = None
|
|
|
|
# Check if current state has expired
|
|
elapsed = now - self._entered_at
|
|
if elapsed < self._duration:
|
|
return self._state
|
|
|
|
# Auto-transition
|
|
next_state = self._next_state()
|
|
self._transition(next_state, now)
|
|
return self._state
|
|
|
|
def on_event(self, event: str, now: float | None = None) -> PipState:
|
|
"""React to a Workshop event.
|
|
|
|
Supported events:
|
|
visitor_entered, visitor_spoke, loud_event, scroll_knocked
|
|
"""
|
|
now = now if now is not None else time.monotonic()
|
|
|
|
if event == "visitor_entered" and self._state in _INTERRUPTIBLE:
|
|
if self._state == PipState.SLEEPING:
|
|
self._transition(PipState.WAKING, now)
|
|
else:
|
|
self._transition(PipState.INVESTIGATING, now)
|
|
|
|
elif event == "visitor_spoke":
|
|
if self._state in (PipState.WANDERING, PipState.WAKING):
|
|
self._transition(PipState.INVESTIGATING, now)
|
|
|
|
elif event == "loud_event":
|
|
if self._state == PipState.SLEEPING:
|
|
self._transition(PipState.WAKING, now)
|
|
|
|
return self._state
|
|
|
|
def on_mood_change(
|
|
self,
|
|
timmy_mood: str,
|
|
confidence: float = 0.5,
|
|
now: float | None = None,
|
|
) -> PipState:
|
|
"""Mirror Timmy's mood with a 3-second delay.
|
|
|
|
Special states triggered by mood + confidence:
|
|
- confidence < 0.3 → ALERT (bristles, particles go red-gold)
|
|
- mood == "energized" → PLAYFUL (figure-8s around crystal ball)
|
|
- mood == "hesitant" + confidence < 0.4 → HIDING
|
|
"""
|
|
now = now if now is not None else time.monotonic()
|
|
|
|
# Schedule mood mirror with 3s delay
|
|
self._pending_mood = timmy_mood
|
|
self._mood_change_at = now + 3.0
|
|
|
|
# Special state triggers (immediate)
|
|
if confidence < 0.3 and self._state in _INTERRUPTIBLE:
|
|
self._transition(PipState.ALERT, now)
|
|
elif timmy_mood == "energized" and self._state in _INTERRUPTIBLE:
|
|
self._transition(PipState.PLAYFUL, now)
|
|
elif timmy_mood == "hesitant" and confidence < 0.4 and self._state in _INTERRUPTIBLE:
|
|
self._transition(PipState.HIDING, now)
|
|
|
|
return self._state
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internals
|
|
# ------------------------------------------------------------------
|
|
|
|
def _transition(self, new_state: PipState, now: float) -> None:
|
|
"""Move to a new state."""
|
|
old = self._state
|
|
self._state = new_state
|
|
self._entered_at = now
|
|
self._duration = random.uniform(*_STATE_DURATIONS[new_state])
|
|
self._position = self._position_for(new_state)
|
|
logger.debug("Pip: %s → %s", old.value, new_state.value)
|
|
|
|
def _next_state(self) -> PipState:
|
|
"""Determine the natural next state after the current one expires."""
|
|
transitions: dict[PipState, PipState] = {
|
|
PipState.SLEEPING: PipState.WAKING,
|
|
PipState.WAKING: PipState.WANDERING,
|
|
PipState.WANDERING: PipState.BORED,
|
|
PipState.INVESTIGATING: PipState.BORED,
|
|
PipState.BORED: PipState.SLEEPING,
|
|
# Special states return to wandering
|
|
PipState.ALERT: PipState.WANDERING,
|
|
PipState.PLAYFUL: PipState.WANDERING,
|
|
PipState.HIDING: PipState.WAKING,
|
|
}
|
|
return transitions.get(self._state, PipState.SLEEPING)
|
|
|
|
def _position_for(self, state: PipState) -> tuple[float, float, float]:
|
|
"""Approximate position hint for a given state.
|
|
|
|
The browser interpolates smoothly; these are target anchors.
|
|
"""
|
|
if state in (PipState.SLEEPING, PipState.BORED):
|
|
return _FIREPLACE_POS
|
|
if state == PipState.HIDING:
|
|
return (0.5, 0.3, -2.0) # Behind the desk
|
|
if state == PipState.PLAYFUL:
|
|
return (1.0, 1.2, 0.0) # Near the crystal ball
|
|
# Wandering / investigating / waking — random room position
|
|
return (
|
|
random.uniform(-1.0, 3.0),
|
|
random.uniform(0.5, 1.5),
|
|
random.uniform(-2.0, 1.0),
|
|
)
|
|
|
|
|
|
# Module-level singleton
|
|
pip_familiar = Familiar()
|