diff --git a/src/timmy/adapters/time_adapter.py b/src/timmy/adapters/time_adapter.py new file mode 100644 index 00000000..697c8496 --- /dev/null +++ b/src/timmy/adapters/time_adapter.py @@ -0,0 +1,82 @@ +"""Time adapter — circadian awareness for Timmy. + +Emits time-of-day events so Timmy knows the current period +and tracks how long since the last user interaction. +""" + +import logging +from datetime import UTC, datetime + +from infrastructure.events.bus import emit + +logger = logging.getLogger(__name__) + +# Time-of-day periods: (event_name, start_hour, end_hour) +_PERIODS = [ + ("morning", 6, 9), + ("afternoon", 12, 14), + ("evening", 18, 20), + ("late_night", 23, 24), + ("late_night", 0, 3), +] + + +def classify_period(hour: int) -> str | None: + """Return the circadian period name for a given hour, or None.""" + for name, start, end in _PERIODS: + if start <= hour < end: + return name + return None + + +class TimeAdapter: + """Emits circadian and interaction-tracking events.""" + + def __init__(self) -> None: + self._last_interaction: datetime | None = None + self._last_period: str | None = None + self._last_date: str | None = None + + def record_interaction(self, now: datetime | None = None) -> None: + """Record a user interaction timestamp.""" + self._last_interaction = now or datetime.now(UTC) + + def time_since_last_interaction( + self, + now: datetime | None = None, + ) -> float | None: + """Seconds since last user interaction, or None if no interaction.""" + if self._last_interaction is None: + return None + current = now or datetime.now(UTC) + return (current - self._last_interaction).total_seconds() + + async def tick(self, now: datetime | None = None) -> list[str]: + """Check current time and emit relevant events. + + Returns list of event types emitted (useful for testing). + """ + current = now or datetime.now(UTC) + emitted: list[str] = [] + + # --- new_day --- + date_str = current.strftime("%Y-%m-%d") + if self._last_date is not None and date_str != self._last_date: + event_type = "time.new_day" + await emit(event_type, source="time_adapter", data={"date": date_str}) + emitted.append(event_type) + self._last_date = date_str + + # --- circadian period --- + period = classify_period(current.hour) + if period is not None and period != self._last_period: + event_type = f"time.{period}" + await emit( + event_type, + source="time_adapter", + data={"hour": current.hour, "period": period}, + ) + emitted.append(event_type) + self._last_period = period + + return emitted diff --git a/tests/timmy/adapters/test_time_adapter.py b/tests/timmy/adapters/test_time_adapter.py new file mode 100644 index 00000000..19fadf37 --- /dev/null +++ b/tests/timmy/adapters/test_time_adapter.py @@ -0,0 +1,146 @@ +"""Tests for the time adapter — circadian awareness.""" + +from datetime import UTC, datetime +from unittest.mock import AsyncMock, patch + +import pytest + +from timmy.adapters.time_adapter import TimeAdapter, classify_period + +# ---------- classify_period ---------- + + +@pytest.mark.parametrize( + "hour, expected", + [ + (6, "morning"), + (7, "morning"), + (8, "morning"), + (9, None), + (12, "afternoon"), + (13, "afternoon"), + (14, None), + (18, "evening"), + (19, "evening"), + (20, None), + (23, "late_night"), + (0, "late_night"), + (2, "late_night"), + (3, None), + (10, None), + (16, None), + ], +) +def test_classify_period(hour: int, expected: str | None) -> None: + assert classify_period(hour) == expected + + +# ---------- record_interaction / time_since ---------- + + +def test_time_since_last_interaction_none() -> None: + adapter = TimeAdapter() + assert adapter.time_since_last_interaction() is None + + +def test_time_since_last_interaction() -> None: + adapter = TimeAdapter() + t0 = datetime(2026, 3, 18, 10, 0, 0, tzinfo=UTC) + t1 = datetime(2026, 3, 18, 10, 5, 0, tzinfo=UTC) + adapter.record_interaction(now=t0) + assert adapter.time_since_last_interaction(now=t1) == 300.0 + + +# ---------- tick — circadian events ---------- + + +@pytest.mark.asyncio +async def test_tick_emits_morning() -> None: + adapter = TimeAdapter() + now = datetime(2026, 3, 18, 7, 0, 0, tzinfo=UTC) + + with patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock) as mock_emit: + emitted = await adapter.tick(now=now) + + assert "time.morning" in emitted + mock_emit.assert_any_call( + "time.morning", + source="time_adapter", + data={"hour": 7, "period": "morning"}, + ) + + +@pytest.mark.asyncio +async def test_tick_emits_late_night() -> None: + adapter = TimeAdapter() + now = datetime(2026, 3, 19, 1, 0, 0, tzinfo=UTC) + + with patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock) as mock_emit: + emitted = await adapter.tick(now=now) + + assert "time.late_night" in emitted + mock_emit.assert_any_call( + "time.late_night", + source="time_adapter", + data={"hour": 1, "period": "late_night"}, + ) + + +@pytest.mark.asyncio +async def test_tick_no_duplicate_period() -> None: + """Same period on consecutive ticks should not re-emit.""" + adapter = TimeAdapter() + t1 = datetime(2026, 3, 18, 7, 0, 0, tzinfo=UTC) + t2 = datetime(2026, 3, 18, 7, 30, 0, tzinfo=UTC) + + with patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock): + await adapter.tick(now=t1) + emitted = await adapter.tick(now=t2) + + assert emitted == [] + + +@pytest.mark.asyncio +async def test_tick_no_event_outside_periods() -> None: + adapter = TimeAdapter() + now = datetime(2026, 3, 18, 10, 0, 0, tzinfo=UTC) + + with patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock) as mock_emit: + emitted = await adapter.tick(now=now) + + assert emitted == [] + mock_emit.assert_not_called() + + +# ---------- tick — new_day ---------- + + +@pytest.mark.asyncio +async def test_tick_emits_new_day() -> None: + adapter = TimeAdapter() + day1 = datetime(2026, 3, 18, 23, 30, 0, tzinfo=UTC) + day2 = datetime(2026, 3, 19, 0, 30, 0, tzinfo=UTC) + + with patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock) as mock_emit: + await adapter.tick(now=day1) + emitted = await adapter.tick(now=day2) + + assert "time.new_day" in emitted + mock_emit.assert_any_call( + "time.new_day", + source="time_adapter", + data={"date": "2026-03-19"}, + ) + + +@pytest.mark.asyncio +async def test_tick_no_new_day_same_date() -> None: + adapter = TimeAdapter() + t1 = datetime(2026, 3, 18, 10, 0, 0, tzinfo=UTC) + t2 = datetime(2026, 3, 18, 15, 0, 0, tzinfo=UTC) + + with patch("timmy.adapters.time_adapter.emit", new_callable=AsyncMock): + await adapter.tick(now=t1) + emitted = await adapter.tick(now=t2) + + assert "time.new_day" not in emitted