Coverage for src\context_awareness.py: 0%

1""" 

2Context Awareness Module - 情境感知系统 

3 

4提供上下文收集、场景识别、环境状态监测功能 

5""" 

6 

7import asyncio 

8import time 

9import platform 

10import psutil 

11from typing import Dict, List, Optional, Any 

12from dataclasses import dataclass, field 

13from enum import Enum 

14from collections import defaultdict 

15import logging 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20class ContextType(Enum): 

21 """上下文类型""" 

22 DEVICE = "device" # 设备信息 

23 TIME = "time" # 时间信息 

24 LOCATION = "location" # 位置信息 

25 ACTIVITY = "activity" # 活动状态 

26 EMOTIONAL = "emotional" # 情绪状态 

27 ENVIRONMENT = "environment" # 环境状态 

28 

29 

30@dataclass 

31class ContextData: 

32 """上下文数据""" 

33 context_type: ContextType 

34 data: Dict[str, Any] 

35 timestamp: float = field(default_factory=time.time) 

36 confidence: float = 1.0 # 置信度 0-1 

37 

38 

39@dataclass 

40class Scene: 

41 """场景""" 

42 scene_id: str 

43 name: str 

44 description: str 

45 context_requirements: Dict[ContextType, float] # 需要的上下文类型和最低置信度 

46 metadata: Dict[str, Any] = field(default_factory=dict) 

47 

48 

49class DeviceContextCollector: 

50 """设备上下文收集器""" 

51 

52 def __init__(self): 

53 self._device_info: Dict[str, Any] = {} 

54 self._system_info: Dict[str, Any] = {} 

55 

56 async def collect(self) -> ContextData: 

57 """收集设备上下文""" 

58 try: 

59 # 基础设备信息 

60 self._device_info = { 

61 "platform": platform.system(), 

62 "platform_version": platform.version(), 

63 "machine": platform.machine(), 

64 "processor": platform.processor(), 

65 } 

66 

67 # 系统资源 

68 self._system_info = { 

69 "cpu_percent": psutil.cpu_percent(interval=0.1), 

70 "memory_percent": psutil.virtual_memory().percent, 

71 "memory_available_mb": psutil.virtual_memory().available // (1024 * 1024), 

72 "disk_percent": psutil.disk_usage('/').percent, 

73 } 

74 

75 # 电池状态（如果可用） 

76 try: 

77 battery = psutil.sensors_battery() 

78 if battery: 

79 self._system_info["battery_percent"] = battery.percent 

80 self._system_info["battery_charging"] = battery.power_plugged 

81 except Exception: 

82 pass 

83 

84 return ContextData( 

85 context_type=ContextType.DEVICE, 

86 data={ 

87 "device": self._device_info, 

88 "system": self._system_info 

89 } 

90 ) 

91 except Exception as e: 

92 logger.error(f"设备上下文收集失败: {e}") 

93 return ContextData( 

94 context_type=ContextType.DEVICE, 

95 data={"error": str(e)} 

96 ) 

97 

98 async def get_device_id(self) -> str: 

99 """获取设备ID""" 

100 return f"{platform.node()}_{platform.machine()}" 

101 

102 

103class TimeContextCollector: 

104 """时间上下文收集器""" 

105 

106 def __init__(self): 

107 self._last_update = 0 

108 

109 async def collect(self) -> ContextData: 

110 """收集时间上下文""" 

111 now = time.time() 

112 local = time.localtime(now) 

113 

114 # 计算时间段 

115 hour = local.tm_hour 

116 if 6 <= hour < 9: 

117 period = "morning" 

118 elif 9 <= hour < 12: 

119 period = "forenoon" 

120 elif 12 <= hour < 14: 

121 period = "noon" 

122 elif 14 <= hour < 18: 

123 period = "afternoon" 

124 elif 18 <= hour < 22: 

125 period = "evening" 

126 else: 

127 period = "night" 

128 

129 # 计算星期 

130 weekdays = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"] 

131 weekday = weekdays[local.tm_wday] 

132 

133 # 判断是否工作日 

134 is_workday = local.tm_wday < 5 

135 

136 return ContextData( 

137 context_type=ContextType.TIME, 

138 data={ 

139 "timestamp": now, 

140 "hour": hour, 

141 "minute": local.tm_min, 

142 "period": period, 

143 "weekday": weekday, 

144 "is_workday": is_workday, 

145 "date": time.strftime("%Y-%m-%d", local) 

146 } 

147 ) 

148 

149 

150class ActivityContextCollector: 

151 """活动上下文收集器""" 

152 

153 def __init__(self): 

154 self._current_activity: str = "idle" 

155 self._activity_history: List[Dict] = [] 

156 self._focus_sessions: List[Dict] = [] 

157 

158 async def collect(self) -> ContextData: 

159 """收集活动上下文""" 

160 return ContextData( 

161 context_type=ContextType.ACTIVITY, 

162 data={ 

163 "current_activity": self._current_activity, 

164 "recent_activities": self._activity_history[-5:], 

165 "focus_sessions_today": len(self._focus_sessions) 

166 } 

167 ) 

168 

169 async def set_activity(self, activity: str) -> None: 

170 """设置当前活动""" 

171 self._current_activity = activity 

172 self._activity_history.append({ 

173 "activity": activity, 

174 "timestamp": time.time() 

175 }) 

176 

177 # 保持历史记录在合理范围 

178 if len(self._activity_history) > 100: 

179 self._activity_history = self._activity_history[-50:] 

180 

181 async def start_focus_session(self) -> None: 

182 """开始专注会话""" 

183 self._focus_sessions.append({ 

184 "start": time.time(), 

185 "end": None 

186 }) 

187 

188 async def end_focus_session(self) -> None: 

189 """结束专注会话""" 

190 if self._focus_sessions and self._focus_sessions[-1]["end"] is None: 

191 self._focus_sessions[-1]["end"] = time.time() 

192 

193 

194class EnvironmentContextCollector: 

195 """环境上下文收集器""" 

196 

197 def __init__(self): 

198 self._network_status = "unknown" 

199 self._screen_status = "unknown" 

200 

201 async def collect(self) -> ContextData: 

202 """收集环境上下文""" 

203 return ContextData( 

204 context_type=ContextType.ENVIRONMENT, 

205 data={ 

206 "network_status": self._network_status, 

207 "screen_status": self._screen_status, 

208 "noise_level": "quiet" # 简化版本 

209 } 

210 ) 

211 

212 async def set_network_status(self, status: str) -> None: 

213 """设置网络状态""" 

214 self._network_status = status 

215 

216 async def set_screen_status(self, status: str) -> None: 

217 """设置屏幕状态""" 

218 self._screen_status = status 

219 

220 

221class SceneRecognizer: 

222 """场景识别器""" 

223 

224 def __init__(self): 

225 self._scenes: Dict[str, Scene] = {} 

226 self._current_scene: Optional[str] = None 

227 self._context_history: List[ContextData] = [] 

228 

229 def register_scene(self, scene: Scene) -> None: 

230 """注册场景""" 

231 self._scenes[scene.scene_id] = scene 

232 logger.info(f"场景已注册: {scene.name}") 

233 

234 async def recognize(self, contexts: List[ContextData]) -> Optional[str]: 

235 """识别当前场景""" 

236 best_match = None 

237 best_score = 0.0 

238 

239 for scene_id, scene in self._scenes.items(): 

240 score = self._calculate_scene_match(scene, contexts) 

241 if score > best_score and score >= 0.5: 

242 best_score = score 

243 best_match = scene_id 

244 

245 if best_match != self._current_scene: 

246 old_scene = self._current_scene 

247 self._current_scene = best_match 

248 logger.info(f"场景切换: {old_scene} -> {best_match} (置信度: {best_score:.2f})") 

249 

250 return self._current_scene 

251 

252 def _calculate_scene_match(self, scene: Scene, contexts: List[ContextData]) -> float: 

253 """计算场景匹配度""" 

254 if not scene.context_requirements: 

255 return 0.0 

256 

257 total_score = 0.0 

258 matched_count = 0 

259 

260 for context in contexts: 

261 req_confidence = scene.context_requirements.get(context.context_type, 0) 

262 if req_confidence > 0 and context.confidence >= req_confidence: 

263 total_score += context.confidence 

264 matched_count += 1 

265 

266 if not scene.context_requirements: 

267 return 0.0 

268 

269 return total_score / len(scene.context_requirements) if matched_count > 0 else 0.0 

270 

271 def get_current_scene(self) -> Optional[Scene]: 

272 """获取当前场景""" 

273 if self._current_scene: 

274 return self._scenes.get(self._current_scene) 

275 return None 

276 

277 

278class ContextAwareness: 

279 """情境感知系统（整合模块）""" 

280 

281 def __init__(self): 

282 self.device_collector = DeviceContextCollector() 

283 self.time_collector = TimeContextCollector() 

284 self.activity_collector = ActivityContextCollector() 

285 self.environment_collector = EnvironmentContextCollector() 

286 self.scene_recognizer = SceneRecognizer() 

287 

288 self._collectors = [ 

289 self.device_collector, 

290 self.time_collector, 

291 self.activity_collector, 

292 self.environment_collector 

293 ] 

294 

295 self._context_cache: List[ContextData] = [] 

296 self._last_collection = 0 

297 self._collection_interval = 60 # 秒 

298 

299 async def collect_all(self, force: bool = False) -> List[ContextData]: 

300 """收集所有上下文""" 

301 now = time.time() 

302 

303 if not force and now - self._last_collection < self._collection_interval: 

304 return self._context_cache 

305 

306 contexts = [] 

307 for collector in self._collectors: 

308 try: 

309 context = await collector.collect() 

310 contexts.append(context) 

311 except Exception as e: 

312 logger.error(f"上下文收集失败: {collector.__class__.__name__}, {e}") 

313 

314 self._context_cache = contexts 

315 self._last_collection = now 

316 

317 return contexts 

318 

319 async def recognize_scene(self) -> Optional[str]: 

320 """识别当前场景""" 

321 contexts = await self.collect_all() 

322 return await self.scene_recognizer.recognize(contexts) 

323 

324 async def register_default_scenes(self) -> None: 

325 """注册默认场景""" 

326 default_scenes = [ 

327 Scene( 

328 scene_id="work_morning", 

329 name="工作日上午", 

330 description="工作日上午专注工作", 

331 context_requirements={ 

332 ContextType.TIME: 0.8, 

333 ContextType.ACTIVITY: 0.5 

334 } 

335 ), 

336 Scene( 

337 scene_id="work_afternoon", 

338 name="工作日下午", 

339 description="工作日下午继续工作", 

340 context_requirements={ 

341 ContextType.TIME: 0.8, 

342 ContextType.ACTIVITY: 0.5 

343 } 

344 ), 

345 Scene( 

346 scene_id="coding", 

347 name="编程模式", 

348 description="专注于编程工作", 

349 context_requirements={ 

350 ContextType.DEVICE: 0.6, 

351 ContextType.ACTIVITY: 0.7 

352 } 

353 ), 

354 Scene( 

355 scene_id="relaxed", 

356 name="休闲放松", 

357 description="非工作时间的休闲状态", 

358 context_requirements={ 

359 ContextType.TIME: 0.9, 

360 ContextType.ENVIRONMENT: 0.3 

361 } 

362 ), 

363 ] 

364 

365 for scene in default_scenes: 

366 self.scene_recognizer.register_scene(scene) 

367 

368 async def summarize_context(self) -> Dict[str, Any]: 

369 """获取上下文摘要""" 

370 contexts = await self.collect_all() 

371 

372 summary = {} 

373 for context in contexts: 

374 summary[context.context_type.value] = context.data 

375 

376 current_scene = self.scene_recognizer.get_current_scene() 

377 if current_scene: 

378 summary["current_scene"] = { 

379 "name": current_scene.name, 

380 "description": current_scene.description 

381 } 

382 

383 return summary