environment.py 1.9 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. """
  2. EnvironmentState - 全局环境状态
  3. 存储环境相关的状态信息,如气体浓度、天气、区域状态等
  4. """
  5. from __future__ import annotations
  6. from dataclasses import dataclass, field
  7. from typing import Any
  8. import time
  9. @dataclass
  10. class EnvironmentState:
  11. """
  12. 环境状态类
  13. Attributes:
  14. timestamp: 状态时间戳
  15. data: 环境数据字典 (如 gas_readings, weather, zones, etc.)
  16. """
  17. timestamp: float = field(default_factory=time.time)
  18. data: dict = field(default_factory=dict)
  19. def update(self, new_data: dict, current_time: float | None = None) -> dict:
  20. """
  21. 更新环境数据,返回变化部分
  22. Args:
  23. new_data: 新数据字典
  24. current_time: 当前时间戳
  25. Returns:
  26. 变化的数据字典
  27. """
  28. if current_time is None:
  29. current_time = time.time()
  30. diff = {}
  31. for key, value in new_data.items():
  32. old_value = self.data.get(key)
  33. if old_value != value:
  34. diff[key] = {'old': old_value, 'new': value}
  35. self.data[key] = value
  36. self.timestamp = current_time
  37. return diff
  38. def get(self, key: str, default: Any = None) -> Any:
  39. """获取环境数据"""
  40. return self.data.get(key, default)
  41. def set(self, key: str, value: Any) -> None:
  42. """设置环境数据"""
  43. self.data[key] = value
  44. self.timestamp = time.time()
  45. def to_dict(self) -> dict:
  46. """转换为字典"""
  47. return {
  48. 'timestamp': self.timestamp,
  49. 'data': self.data,
  50. }
  51. @classmethod
  52. def from_dict(cls, data: dict) -> EnvironmentState:
  53. """从字典创建"""
  54. return cls(
  55. timestamp=data.get('timestamp', time.time()),
  56. data=data.get('data', {}),
  57. )