forked from Rockachopa/Timmy-time-dashboard
@@ -5,9 +5,45 @@ into the camelCase world-state payload consumed by the Workshop 3D renderer
|
||||
and WebSocket gateway.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Default Pip familiar state (used when familiar module unavailable)
|
||||
DEFAULT_PIP_STATE = {
|
||||
"name": "Pip",
|
||||
"mood": "sleepy",
|
||||
"energy": 0.5,
|
||||
"color": "0x00b450", # emerald green
|
||||
"trail_color": "0xdaa520", # gold
|
||||
}
|
||||
|
||||
|
||||
def _get_familiar_state() -> dict:
|
||||
"""Get Pip familiar state from familiar module, with graceful fallback.
|
||||
|
||||
Returns a dict with name, mood, energy, color, and trail_color.
|
||||
Falls back to default state if familiar module unavailable or raises.
|
||||
"""
|
||||
try:
|
||||
from timmy.familiar import pip_familiar
|
||||
|
||||
snapshot = pip_familiar.snapshot()
|
||||
# Map PipSnapshot fields to the expected agent_state format
|
||||
return {
|
||||
"name": snapshot.name,
|
||||
"mood": snapshot.state,
|
||||
"energy": DEFAULT_PIP_STATE["energy"], # Pip doesn't track energy yet
|
||||
"color": DEFAULT_PIP_STATE["color"],
|
||||
"trail_color": DEFAULT_PIP_STATE["trail_color"],
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.warning("Familiar state unavailable, using default: %s", exc)
|
||||
return DEFAULT_PIP_STATE.copy()
|
||||
|
||||
|
||||
# Valid bark styles for Matrix protocol
|
||||
BARK_STYLES = {"speech", "thought", "whisper", "shout"}
|
||||
|
||||
@@ -200,6 +236,7 @@ def produce_agent_state(agent_id: str, presence: dict) -> dict:
|
||||
"mood": presence.get("mood", "calm"),
|
||||
"energy": presence.get("energy", 0.5),
|
||||
"bark": presence.get("bark", ""),
|
||||
"familiar": _get_familiar_state(),
|
||||
},
|
||||
"ts": int(time.time()),
|
||||
}
|
||||
|
||||
@@ -5,6 +5,8 @@ from unittest.mock import patch
|
||||
import pytest
|
||||
|
||||
from infrastructure.presence import (
|
||||
DEFAULT_PIP_STATE,
|
||||
_get_familiar_state,
|
||||
produce_agent_state,
|
||||
produce_bark,
|
||||
produce_thought,
|
||||
@@ -169,6 +171,66 @@ class TestProduceAgentState:
|
||||
data = produce_agent_state("spark", {})["data"]
|
||||
assert data["display_name"] == "Spark"
|
||||
|
||||
def test_familiar_in_data(self):
|
||||
"""agent_state.data includes familiar field with required keys."""
|
||||
data = produce_agent_state("timmy", {})["data"]
|
||||
|
||||
assert "familiar" in data
|
||||
familiar = data["familiar"]
|
||||
assert familiar["name"] == "Pip"
|
||||
assert "mood" in familiar
|
||||
assert "energy" in familiar
|
||||
assert familiar["color"] == "0x00b450"
|
||||
assert familiar["trail_color"] == "0xdaa520"
|
||||
|
||||
def test_familiar_has_all_required_fields(self):
|
||||
"""familiar dict contains all required fields per acceptance criteria."""
|
||||
data = produce_agent_state("timmy", {})["data"]
|
||||
familiar = data["familiar"]
|
||||
|
||||
required_fields = {"name", "mood", "energy", "color", "trail_color"}
|
||||
assert set(familiar.keys()) >= required_fields
|
||||
|
||||
|
||||
class TestFamiliarState:
|
||||
"""Tests for _get_familiar_state() — Pip familiar state retrieval."""
|
||||
|
||||
def test_get_familiar_state_returns_dict(self):
|
||||
"""_get_familiar_state returns a dict."""
|
||||
result = _get_familiar_state()
|
||||
assert isinstance(result, dict)
|
||||
|
||||
def test_get_familiar_state_has_required_fields(self):
|
||||
"""Result contains name, mood, energy, color, trail_color."""
|
||||
result = _get_familiar_state()
|
||||
|
||||
assert result["name"] == "Pip"
|
||||
assert "mood" in result
|
||||
assert isinstance(result["energy"], (int, float))
|
||||
assert result["color"] == "0x00b450"
|
||||
assert result["trail_color"] == "0xdaa520"
|
||||
|
||||
def test_default_pip_state_constant(self):
|
||||
"""DEFAULT_PIP_STATE has expected values."""
|
||||
assert DEFAULT_PIP_STATE["name"] == "Pip"
|
||||
assert DEFAULT_PIP_STATE["mood"] == "sleepy"
|
||||
assert DEFAULT_PIP_STATE["energy"] == 0.5
|
||||
assert DEFAULT_PIP_STATE["color"] == "0x00b450"
|
||||
assert DEFAULT_PIP_STATE["trail_color"] == "0xdaa520"
|
||||
|
||||
@patch("infrastructure.presence.logger")
|
||||
def test_get_familiar_state_fallback_on_exception(self, mock_logger):
|
||||
"""When familiar module raises, falls back to default and logs warning."""
|
||||
# Patch inside the function where pip_familiar is imported
|
||||
with patch("timmy.familiar.pip_familiar.snapshot") as mock_snapshot:
|
||||
mock_snapshot.side_effect = RuntimeError("Pip is napping")
|
||||
result = _get_familiar_state()
|
||||
|
||||
assert result["name"] == "Pip"
|
||||
assert result["mood"] == "sleepy"
|
||||
mock_logger.warning.assert_called_once()
|
||||
assert "Pip is napping" in str(mock_logger.warning.call_args)
|
||||
|
||||
|
||||
class TestProduceBark:
|
||||
"""Tests for produce_bark() — Matrix bark message producer."""
|
||||
|
||||
Reference in New Issue
Block a user