feat: time adapter — circadian awareness for Timmy
All checks were successful
Tests / lint (pull_request) Successful in 5s
Tests / test (pull_request) Successful in 1m3s

Add time_adapter that emits time-of-day events (morning, afternoon,
evening, late_night, new_day) and tracks idle time since last user
interaction via the event bus.

Fixes #307

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
kimi
2026-03-18 18:50:21 -04:00
parent 39939270b7
commit c3ed0ac830
2 changed files with 258 additions and 0 deletions

View File

@@ -0,0 +1,112 @@
"""Time adapter — circadian awareness for Timmy.
Emits time-of-day events so Timmy knows the current period (morning,
afternoon, evening, late_night) and tracks time since the last user
interaction. All times use the system's local timezone.
"""
import logging
from datetime import UTC, datetime
from infrastructure.events.bus import emit
logger = logging.getLogger(__name__)
# ── Time period definitions (hour ranges, inclusive start, exclusive end) ─────
PERIODS: dict[str, list[tuple[int, int]]] = {
"morning": [(6, 9)],
"afternoon": [(12, 14)],
"evening": [(18, 20)],
"late_night": [(23, 24), (0, 3)],
}
EVENT_SOURCE = "time_adapter"
def get_current_period(now: datetime | None = None) -> str | None:
"""Return the circadian period name for the given time, or None."""
if now is None:
now = datetime.now().astimezone()
hour = now.hour
for period, ranges in PERIODS.items():
for start, end in ranges:
if start <= hour < end:
return period
return None
def is_new_day(now: datetime | None = None, previous: datetime | None = None) -> bool:
"""Return True if *now* is on a different calendar date than *previous*."""
if previous is None:
return False
if now is None:
now = datetime.now().astimezone()
return now.date() != previous.date()
def time_since_last_interaction(
last_interaction: datetime | None = None,
now: datetime | None = None,
) -> float | None:
"""Return seconds since the last user interaction, or None."""
if last_interaction is None:
return None
if now is None:
now = datetime.now(UTC)
delta = now - last_interaction
return max(delta.total_seconds(), 0.0)
# ── Emit helpers (called by the app's tick / scheduler) ──────────────────────
async def emit_time_events(
now: datetime | None = None,
previous_tick: datetime | None = None,
last_interaction: datetime | None = None,
) -> list[str]:
"""Check the clock and emit any relevant time events.
Returns a list of event type strings that were emitted (useful for
testing and logging).
"""
if now is None:
now = datetime.now().astimezone()
emitted: list[str] = []
# ── new_day ──────────────────────────────────────────────────────────
if is_new_day(now, previous_tick):
await emit(
"time.new_day",
source=EVENT_SOURCE,
data={"date": now.date().isoformat()},
)
emitted.append("time.new_day")
# ── period transition ────────────────────────────────────────────────
period = get_current_period(now)
prev_period = get_current_period(previous_tick) if previous_tick else None
if period is not None and period != prev_period:
await emit(
f"time.{period}",
source=EVENT_SOURCE,
data={"period": period, "hour": now.hour},
)
emitted.append(f"time.{period}")
# ── idle tracking ────────────────────────────────────────────────────
idle_seconds = time_since_last_interaction(last_interaction, now)
if idle_seconds is not None and idle_seconds > 0:
await emit(
"time.idle",
source=EVENT_SOURCE,
data={"idle_seconds": round(idle_seconds, 1)},
)
emitted.append("time.idle")
if emitted:
logger.debug("Time events emitted: %s", emitted)
return emitted

View File

@@ -0,0 +1,146 @@
"""Tests for the time adapter — circadian awareness."""
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, patch
import pytest
from timmy.adapters.time_adapter import (
emit_time_events,
get_current_period,
is_new_day,
time_since_last_interaction,
)
def _dt(hour: int, minute: int = 0, day: int = 15) -> datetime:
"""Build a timezone-aware datetime for testing."""
return datetime(2026, 3, day, hour, minute, tzinfo=UTC)
# ── get_current_period ───────────────────────────────────────────────────────
class TestGetCurrentPeriod:
@pytest.mark.parametrize(
"hour,expected",
[
(6, "morning"),
(7, "morning"),
(8, "morning"),
(12, "afternoon"),
(13, "afternoon"),
(18, "evening"),
(19, "evening"),
(23, "late_night"),
(0, "late_night"),
(1, "late_night"),
(2, "late_night"),
],
)
def test_known_periods(self, hour, expected):
assert get_current_period(_dt(hour)) == expected
@pytest.mark.parametrize("hour", [5, 9, 10, 11, 14, 15, 16, 17, 20, 21, 22])
def test_no_period(self, hour):
assert get_current_period(_dt(hour)) is None
# ── is_new_day ───────────────────────────────────────────────────────────────
class TestIsNewDay:
def test_same_day(self):
assert is_new_day(_dt(1, day=15), _dt(23, day=15)) is False
def test_different_day(self):
assert is_new_day(_dt(0, day=16), _dt(23, day=15)) is True
def test_no_previous(self):
assert is_new_day(_dt(0), None) is False
# ── time_since_last_interaction ──────────────────────────────────────────────
class TestTimeSinceLastInteraction:
def test_returns_seconds(self):
last = _dt(10)
now = _dt(10) + timedelta(minutes=5)
assert time_since_last_interaction(last, now) == 300.0
def test_none_when_no_interaction(self):
assert time_since_last_interaction(None, _dt(10)) is None
def test_never_negative(self):
last = _dt(10, 30)
now = _dt(10, 0)
assert time_since_last_interaction(last, now) == 0.0
# ── emit_time_events ────────────────────────────────────────────────────────
@pytest.mark.asyncio
class TestEmitTimeEvents:
@patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock)
async def test_morning_emitted(self, mock_emit):
now = _dt(7)
prev = _dt(5) # hour 5 → no period
result = await emit_time_events(now=now, previous_tick=prev)
assert "time.morning" in result
# find the morning call
calls = [c for c in mock_emit.call_args_list if c[0][0] == "time.morning"]
assert len(calls) == 1
assert calls[0][1]["data"]["period"] == "morning"
@patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock)
async def test_new_day_emitted(self, mock_emit):
prev = _dt(22, day=15) # hour 22 = no period
now = _dt(0, day=16) # hour 0 = late_night + new day
result = await emit_time_events(now=now, previous_tick=prev)
assert "time.new_day" in result
assert "time.late_night" in result
@patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock)
async def test_no_period_no_emit(self, mock_emit):
"""Hour 10 is not in any period — only idle should fire."""
now = _dt(10)
prev = _dt(9)
result = await emit_time_events(now=now, previous_tick=prev)
assert "time.morning" not in result
assert "time.afternoon" not in result
@patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock)
async def test_same_period_no_duplicate(self, mock_emit):
"""Staying in the same period should not re-emit."""
now = _dt(7)
prev = _dt(6) # both morning
result = await emit_time_events(now=now, previous_tick=prev)
assert "time.morning" not in result
@patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock)
async def test_idle_emitted(self, mock_emit):
now = _dt(10)
last_interaction = _dt(10) - timedelta(minutes=30)
result = await emit_time_events(
now=now,
previous_tick=_dt(9),
last_interaction=last_interaction,
)
assert "time.idle" in result
idle_call = [c for c in mock_emit.call_args_list if c[0][0] == "time.idle"]
assert idle_call[0][1]["data"]["idle_seconds"] == 1800.0
@patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock)
async def test_no_idle_without_interaction(self, mock_emit):
result = await emit_time_events(now=_dt(10), previous_tick=_dt(9))
assert "time.idle" not in result
@patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock)
async def test_late_night_spans_midnight(self, mock_emit):
"""late_night covers 23-24 and 0-3."""
for hour in (23, 0, 1, 2):
mock_emit.reset_mock()
result = await emit_time_events(now=_dt(hour), previous_tick=_dt(22))
assert "time.late_night" in result