""" Entity - 实体抽象类 支持状态 Diff、TTL 过期、metadata 扩展 """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any import time @dataclass class Entity: """ 实体抽象类 Attributes: entity_id: 实体唯一标识 entity_type: 实体类型 (animal, obstacle, person, actuator, etc.) state: 当前状态字典 metadata: 元数据字典 (source, confidence, timestamp, etc.) last_update: 最后更新时间戳 ttl: 生存时间 (秒),None 表示永不过期 """ entity_id: str entity_type: str state: dict = field(default_factory=dict) metadata: dict | None = None last_update: float = field(default_factory=time.time) ttl: float | None = None def is_expired(self, current_time: float | None = None) -> bool: """检查实体是否已过期""" if self.ttl is None: return False if current_time is None: current_time = time.time() return (current_time - self.last_update) > self.ttl def update_state(self, new_state: dict, current_time: float | None = None) -> dict: """ 更新实体状态,返回状态变化 (Diff) Args: new_state: 新状态字典 current_time: 当前时间戳 Returns: 状态变化字典 """ if current_time is None: current_time = time.time() diff = {} for key, value in new_state.items(): old_value = self.state.get(key) if old_value != value: diff[key] = {'old': old_value, 'new': value} self.state[key] = value self.last_update = current_time return diff def update_metadata(self, metadata: dict) -> None: """更新元数据""" if self.metadata is None: self.metadata = {} self.metadata.update(metadata) def to_dict(self) -> dict: """转换为字典""" return { 'id': self.entity_id, 'type': self.entity_type, 'state': self.state, 'metadata': self.metadata, 'last_update': self.last_update, 'ttl': self.ttl, } @classmethod def from_dict(cls, data: dict) -> Entity: """从字典创建实体""" return cls( entity_id=data['id'], entity_type=data['type'], state=data.get('state', {}), metadata=data.get('metadata'), last_update=data.get('last_update', time.time()), ttl=data.get('ttl'), )