Coverage for src\core\device.py: 0%

1""" 

2设备模型定义 

3定义 PAO 系统中的设备实体和状态 

4""" 

5 

6import uuid 

7import socket 

8from datetime import datetime 

9from typing import Dict, Any, Optional, List 

10from enum import Enum 

11from dataclasses import dataclass, field 

12from pydantic import BaseModel, Field 

13 

14 

15class DeviceType(str, Enum): 

16 """设备类型枚举""" 

17 DESKTOP = "desktop" 

18 LAPTOP = "laptop" 

19 PHONE = "phone" 

20 TABLET = "tablet" 

21 SERVER = "server" 

22 IOT = "iot" 

23 UNKNOWN = "unknown" 

24 

25 

26class DeviceStatus(str, Enum): 

27 """设备状态枚举""" 

28 ONLINE = "online" 

29 OFFLINE = "offline" 

30 BUSY = "busy" 

31 IDLE = "idle" 

32 SLEEPING = "sleeping" 

33 

34 

35class DeviceCapability(str, Enum): 

36 """设备能力枚举""" 

37 COMPUTE = "compute" # 计算能力 

38 STORAGE = "storage" # 存储能力 

39 MEMORY = "memory" # 记忆存储 

40 DISPLAY = "display" # 显示能力 

41 INPUT = "input" # 输入能力 

42 NETWORK = "network" # 网络能力 

43 SENSORS = "sensors" # 传感器能力 

44 

45 

46@dataclass 

47class DeviceInfo: 

48 """设备基本信息""" 

49 device_id: str = field(default_factory=lambda: str(uuid.uuid4())) 

50 name: str = "Unknown Device" 

51 device_type: DeviceType = DeviceType.UNKNOWN 

52 hostname: str = field(default_factory=socket.gethostname) 

53 ip_address: str = field(default_factory=lambda: socket.gethostbyname(socket.gethostname())) 

54 mac_address: str = "" 

55 os_name: str = "" 

56 os_version: str = "" 

57 

58 # 设备能力 

59 capabilities: List[DeviceCapability] = field(default_factory=list) 

60 

61 # 资源信息 

62 cpu_cores: int = 0 

63 memory_gb: int = 0 

64 storage_gb: int = 0 

65 

66 # 网络信息 

67 port: int = 8765 # 默认通信端口 

68 protocol_version: str = "1.0.0" 

69 

70 # 元数据 

71 first_seen: datetime = field(default_factory=datetime.now) 

72 last_seen: datetime = field(default_factory=datetime.now) 

73 

74 def to_dict(self) -> Dict[str, Any]: 

75 """转换为字典""" 

76 return { 

77 "device_id": self.device_id, 

78 "name": self.name, 

79 "device_type": self.device_type.value, 

80 "hostname": self.hostname, 

81 "ip_address": self.ip_address, 

82 "mac_address": self.mac_address, 

83 "os_name": self.os_name, 

84 "os_version": self.os_version, 

85 "capabilities": [cap.value for cap in self.capabilities], 

86 "cpu_cores": self.cpu_cores, 

87 "memory_gb": self.memory_gb, 

88 "storage_gb": self.storage_gb, 

89 "port": self.port, 

90 "protocol_version": self.protocol_version, 

91 "first_seen": self.first_seen.isoformat(), 

92 "last_seen": self.last_seen.isoformat() 

93 } 

94 

95 @classmethod 

96 def from_dict(cls, data: Dict[str, Any]) -> "DeviceInfo": 

97 """从字典创建""" 

98 device = cls() 

99 device.device_id = data.get("device_id", device.device_id) 

100 device.name = data.get("name", device.name) 

101 device.device_type = DeviceType(data.get("device_type", DeviceType.UNKNOWN.value)) 

102 device.hostname = data.get("hostname", device.hostname) 

103 device.ip_address = data.get("ip_address", device.ip_address) 

104 device.mac_address = data.get("mac_address", device.mac_address) 

105 device.os_name = data.get("os_name", device.os_name) 

106 device.os_version = data.get("os_version", device.os_version) 

107 

108 # 处理能力列表 

109 capabilities = data.get("capabilities", []) 

110 device.capabilities = [DeviceCapability(cap) for cap in capabilities] 

111 

112 device.cpu_cores = data.get("cpu_cores", device.cpu_cores) 

113 device.memory_gb = data.get("memory_gb", device.memory_gb) 

114 device.storage_gb = data.get("storage_gb", device.storage_gb) 

115 device.port = data.get("port", device.port) 

116 device.protocol_version = data.get("protocol_version", device.protocol_version) 

117 

118 # 时间戳 

119 if "first_seen" in data: 

120 device.first_seen = datetime.fromisoformat(data["first_seen"]) 

121 if "last_seen" in data: 

122 device.last_seen = datetime.fromisoformat(data["last_seen"]) 

123 

124 return device 

125 

126 def update_last_seen(self): 

127 """更新最后出现时间""" 

128 self.last_seen = datetime.now() 

129 

130 def get_capability_score(self, capability: DeviceCapability) -> int: 

131 """获取设备特定能力的评分""" 

132 if capability not in self.capabilities: 

133 return 0 

134 

135 # 根据能力类型和设备规格打分 

136 scores = { 

137 DeviceCapability.COMPUTE: self.cpu_cores * 100, 

138 DeviceCapability.STORAGE: self.storage_gb * 10, 

139 DeviceCapability.MEMORY: self.memory_gb * 50, 

140 DeviceCapability.DISPLAY: 100 if DeviceCapability.DISPLAY in self.capabilities else 0, 

141 DeviceCapability.INPUT: 80 if DeviceCapability.INPUT in self.capabilities else 0, 

142 DeviceCapability.NETWORK: 150, 

143 DeviceCapability.SENSORS: 50 if DeviceCapability.SENSORS in self.capabilities else 0, 

144 } 

145 

146 return scores.get(capability, 0) 

147 

148 

149class DeviceState(BaseModel): 

150 """设备状态信息""" 

151 device_id: str 

152 status: DeviceStatus = DeviceStatus.ONLINE 

153 current_load: float = 0.0 # 0-1 表示负载程度 

154 available_memory_mb: int = 0 

155 available_storage_mb: int = 0 

156 network_latency_ms: int = 0 

157 active_connections: int = 0 

158 last_heartbeat: datetime = Field(default_factory=datetime.now) 

159 

160 class Config: 

161 json_encoders = { 

162 datetime: lambda v: v.isoformat() 

163 } 

164 

165 

166class DeviceRegistry: 

167 """设备注册表，管理已知设备""" 

168 

169 def __init__(self): 

170 self.devices: Dict[str, DeviceInfo] = {} 

171 self.device_states: Dict[str, DeviceState] = {} 

172 

173 def register_device(self, device: DeviceInfo) -> None: 

174 """注册新设备""" 

175 self.devices[device.device_id] = device 

176 self.device_states[device.device_id] = DeviceState( 

177 device_id=device.device_id, 

178 status=DeviceStatus.ONLINE 

179 ) 

180 

181 def unregister_device(self, device_id: str) -> None: 

182 """注销设备""" 

183 if device_id in self.devices: 

184 del self.devices[device_id] 

185 if device_id in self.device_states: 

186 del self.device_states[device_id] 

187 

188 def update_device_state(self, device_id: str, state: DeviceState) -> None: 

189 """更新设备状态""" 

190 if device_id in self.device_states: 

191 self.device_states[device_id] = state 

192 

193 def get_device(self, device_id: str) -> Optional[DeviceInfo]: 

194 """获取设备信息""" 

195 return self.devices.get(device_id) 

196 

197 def get_device_state(self, device_id: str) -> Optional[DeviceState]: 

198 """获取设备状态""" 

199 return self.device_states.get(device_id) 

200 

201 def list_devices(self) -> List[DeviceInfo]: 

202 """列出所有设备""" 

203 return list(self.devices.values()) 

204 

205 def list_online_devices(self) -> List[DeviceInfo]: 

206 """列出在线设备""" 

207 online_devices = [] 

208 for device_id, state in self.device_states.items(): 

209 if state.status == DeviceStatus.ONLINE and device_id in self.devices: 

210 online_devices.append(self.devices[device_id]) 

211 return online_devices 

212 

213 def find_device_by_capability(self, capability: DeviceCapability) -> List[DeviceInfo]: 

214 """根据能力查找设备""" 

215 return [ 

216 device for device in self.devices.values() 

217 if capability in device.capabilities 

218 ] 

219 

220 def cleanup_offline_devices(self, timeout_seconds: int = 300) -> None: 

221 """清理离线超时的设备""" 

222 now = datetime.now() 

223 to_remove = [] 

224 

225 for device_id, state in self.device_states.items(): 

226 time_since_heartbeat = (now - state.last_heartbeat).total_seconds() 

227 if time_since_heartbeat > timeout_seconds: 

228 to_remove.append(device_id) 

229 

230 for device_id in to_remove: 

231 self.unregister_device(device_id)