aura/perception.py

"""感知引擎：监测用户状态，构建决策上下文"""

import time
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, List


class DeviceMode(Enum):
    NORMAL = "normal"
    MEETING = "meeting"
    SLEEP = "sleep"
    DRIVING = "driving"
    FOCUS = "focus"


class TimeOfDay(Enum):
    MORNING = "morning"     # 07-12
    AFTERNOON = "afternoon" # 12-18
    EVENING = "evening"     # 18-23
    NIGHT = "night"         # 23-07
    NEUTRAL = "neutral"


class UserMood(Enum):
    HAPPY = "happy"
    SAD = "sad"
    ANXIOUS = "anxious"
    ANGRY = "angry"
    TIRED = "tired"
    NEUTRAL = "neutral"
    UNKNOWN = "unknown"


@dataclass
class UserContext:
    """用户当前上下文快照"""
    user_id: str
    idle_minutes: float = 0.0
    time_of_day: TimeOfDay = TimeOfDay.NEUTRAL
    device_mode: DeviceMode = DeviceMode.NORMAL
    mood: UserMood = UserMood.UNKNOWN
    is_active_session: bool = False
    last_conversation_topic: Optional[str] = None
    upcoming_events: List[str] = field(default_factory=list)
    unread_notifications: int = 0
    location: Optional[str] = None
    heart_rate: Optional[float] = None          # 仅本地传感器
    screen_on: bool = True
    last_interaction_time: float = field(default_factory=time.time)

    def to_dict(self) -> dict:
        return {
            "user_id": self.user_id,
            "idle_minutes": round(self.idle_minutes, 1),
            "time_of_day": self.time_of_day.value,
            "device_mode": self.device_mode.value,
            "mood": self.mood.value,
            "is_active_session": self.is_active_session,
            "topic": self.last_conversation_topic or "",
            "events": self.upcoming_events,
        }


class PerceptionEngine:
    """
    感知引擎：聚合所有输入信号，构建用户状态
    """

    def __init__(self):
        self._last_tick = time.time()
        self._context = UserContext(user_id="default")

    def update(self, **kwargs):
        """更新用户上下文字段"""
        for k, v in kwargs.items():
            if hasattr(self._context, k):
                setattr(self._context, k, v)
        self._context.last_interaction_time = time.time()

    def tick(self) -> UserContext:
        """心跳更新：刷新空闲时间、时段等"""
        now = time.time()
        self._context.idle_minutes = (now - self._context.last_interaction_time) / 60.0
        self._context.time_of_day = self._detect_time_of_day()
        return self._context

    def get_context(self) -> UserContext:
        return self._context

    @staticmethod
    def _detect_time_of_day() -> TimeOfDay:
        h = time.localtime().tm_hour
        if 7 <= h < 12: return TimeOfDay.MORNING
        if 12 <= h < 18: return TimeOfDay.AFTERNOON
        if 18 <= h < 23: return TimeOfDay.EVENING
        return TimeOfDay.NIGHT

    @staticmethod
    def detect_mood_from_text(text: str) -> UserMood:
        """从文本中简单检测情绪"""
        keywords = {
            UserMood.HAPPY: ["开心", "高兴", "哈哈", "太好", "谢谢", "nice", "great", "happy"],
            UserMood.SAD: ["难过", "伤心", "哭", "难受", "sad", "depressed", "失望"],
            UserMood.ANXIOUS: ["焦虑", "紧张", "担心", "害怕", "stress", "anxious", "panic"],
            UserMood.ANGRY: ["生气", "愤怒", "烦", "滚", "angry", "mad", "furious"],
            UserMood.TIRED: ["累", "困", "疲惫", "tired", "exhausted", "sleepy"],
        }
        text_lower = text.lower()
        scores = {}
        for mood, words in keywords.items():
            scores[mood] = sum(1 for w in words if w in text_lower)
        if not any(scores.values()):
            return UserMood.NEUTRAL
        return max(scores, key=scores.get)