BodyBalanceEvaluation/backend/devices/device_coordinator.py

1008 lines
39 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
设备协调器
负责统一管理和协调所有设备的生命周期数据流和状态同步
"""
import threading
import time
import logging
from typing import Dict, List, Optional, Any, Callable
from collections import defaultdict
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
try:
from .camera_manager import CameraManager
from .imu_manager import IMUManager
from .pressure_manager import PressureManager
from .femtobolt_manager import FemtoBoltManager
from .utils.config_manager import ConfigManager
from .utils.socket_manager import SocketManager
except ImportError:
from camera_manager import CameraManager
from imu_manager import IMUManager
from pressure_manager import PressureManager
from femtobolt_manager import FemtoBoltManager
from utils.config_manager import ConfigManager
from utils.socket_manager import SocketManager
class DeviceCoordinator:
"""设备协调器 - 统一管理所有设备"""
def __init__(self, socketio, config_path: Optional[str] = None):
"""
初始化设备协调器
Args:
socketio: SocketIO实例
config_path: 配置文件路径
"""
self.socketio = socketio
self.logger = logging.getLogger(self.__class__.__name__)
# 配置管理
self.config_manager = ConfigManager(config_path)
self.socket_manager = SocketManager(socketio)
# 设备管理器
self.devices: Dict[str, Any] = {}
# 获取设备配置
self.device_configs = self.config_manager.get_all_device_configs()
# 状态管理
self.is_initialized = False
self.is_running = False
self.coordinator_lock = threading.RLock()
# 监控线程
self.monitor_thread = None
self.health_check_interval = 5.0 # 健康检查间隔(秒)
# 事件回调
self.event_callbacks: Dict[str, List[Callable]] = defaultdict(list)
# 状态变化回调存储
self._status_change_callback = None
# 性能统计
self.stats = {
'start_time': None,
'total_frames': 0,
'device_errors': defaultdict(int),
'reconnect_attempts': defaultdict(int)
}
# 线程池
self.executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="DeviceCoord")
self.logger.info("设备协调器初始化完成")
def set_status_change_callback(self, callback: Callable):
"""
设置状态变化回调函数
Args:
callback: 状态变化回调函数
"""
self._status_change_callback = callback
# 为已存在的设备注册回调
for device_name, device in self.devices.items():
if device and hasattr(device, 'add_status_change_callback'):
device.add_status_change_callback(callback)
self.logger.info(f"{device_name} 设备状态变化回调已注册")
def initialize(self) -> bool:
"""
初始化所有设备
Returns:
bool: 初始化是否成功
"""
with self.coordinator_lock:
if self.is_initialized:
self.logger.warning("设备协调器已初始化")
return True
try:
self.logger.info("开始初始化设备协调器...")
# 注册Socket.IO命名空间
self._register_namespaces()
# 初始化设备
if not self._initialize_devices():
self.logger.warning("设备初始化失败,将以降级模式继续运行")
# 启动监控线程
self._start_monitor()
self.is_initialized = True
self.stats['start_time'] = time.time()
self.logger.info("设备协调器初始化成功")
self._emit_event('coordinator_initialized', {'devices': list(self.devices.keys())})
return True
except Exception as e:
self.logger.error(f"设备协调器初始化失败: {e}")
self._cleanup_devices()
return False
def _register_namespaces(self):
"""
注册Socket.IO命名空间
"""
namespace_mappings = {
'/devices': 'devices',
'/coordinator': 'coordinator'
}
for namespace, device_name in namespace_mappings.items():
self.socket_manager.register_namespace(namespace, device_name)
self.logger.info(f"已注册Socket.IO命名空间: {list(namespace_mappings.keys())}")
def _initialize_devices(self) -> bool:
"""
初始化所有设备
Returns:
bool: 初始化是否成功
"""
try:
# 并行初始化设备
futures = []
# FemtoBolt深度相机
if self.device_configs.get('femtobolt', {}).get('enabled', False):
future = self.executor.submit(self._init_femtobolt)
futures.append(('femtobolt', future))
# 普通相机
if self.device_configs.get('camera', {}).get('enabled', False):
future = self.executor.submit(self._init_camera)
futures.append(('camera', future))
# IMU传感器
if self.device_configs.get('imu', {}).get('enabled', False):
future = self.executor.submit(self._init_imu)
futures.append(('imu', future))
# 压力传感器
if self.device_configs.get('pressure', {}).get('enabled', False):
future = self.executor.submit(self._init_pressure)
futures.append(('pressure', future))
# 等待所有设备初始化完成
success_count = 0
for device_name, future in futures:
try:
result = future.result(timeout=30) # 30秒超时
if result:
success_count += 1
self.logger.info(f"{device_name}设备初始化成功")
else:
self.logger.error(f"{device_name}设备初始化失败")
except Exception as e:
self.logger.error(f"{device_name}设备初始化异常: {e}")
# 至少需要一个设备初始化成功
if success_count == 0:
self.logger.warning("没有设备初始化成功,但系统将继续运行")
return False
self.logger.info(f"设备初始化完成,成功: {success_count}/{len(futures)}")
return True
except Exception as e:
self.logger.error(f"设备初始化失败: {e}")
return False
def _init_camera(self) -> bool:
"""
初始化普通相机
Returns:
bool: 初始化是否成功
"""
try:
camera = CameraManager(self.socketio, self.config_manager)
self.devices['camera'] = camera
if camera.initialize():
return True
return False
except Exception as e:
self.logger.error(f"初始化相机失败: {e}")
return False
def _init_imu(self) -> bool:
"""
初始化IMU传感器
Returns:
bool: 初始化是否成功
"""
try:
imu = IMUManager(self.socketio, self.config_manager)
self.devices['imu'] = imu
if imu.initialize():
return True
return False
except Exception as e:
self.logger.error(f"初始化IMU失败: {e}")
return False
def _init_pressure(self) -> bool:
"""
初始化压力传感器
Returns:
bool: 初始化是否成功
"""
try:
pressure = PressureManager(self.socketio, self.config_manager)
self.devices['pressure'] = pressure
if pressure.initialize():
return True
return False
except Exception as e:
self.logger.error(f"初始化压力传感器失败: {e}")
return False
def _init_femtobolt(self) -> bool:
"""
初始化FemtoBolt深度相机
Returns:
bool: 初始化是否成功
"""
try:
femtobolt = FemtoBoltManager(self.socketio, self.config_manager)
self.devices['femtobolt'] = femtobolt
if femtobolt.initialize():
return True
return False
except Exception as e:
self.logger.error(f"初始化FemtoBolt失败: {e}")
return False
def start_all_streaming(self) -> bool:
"""
启动所有设备的数据流
Returns:
bool: 启动是否成功
"""
with self.coordinator_lock:
if not self.is_initialized:
self.logger.error("设备协调器未初始化")
return False
if self.is_running:
self.logger.warning("设备流已在运行")
return True
try:
self.logger.info("启动所有设备数据流...")
# 并行启动所有设备流
futures = []
for device_name, device in self.devices.items():
future = self.executor.submit(device.start_streaming)
futures.append((device_name, future))
# 等待所有设备启动完成
success_count = 0
for device_name, future in futures:
try:
result = future.result(timeout=10) # 10秒超时
if result:
success_count += 1
self.logger.info(f"{device_name}数据流启动成功")
else:
self.logger.error(f"{device_name}数据流启动失败")
except Exception as e:
self.logger.error(f"{device_name}数据流启动异常: {e}")
self.is_running = success_count > 0
if self.is_running:
self.logger.info(f"设备数据流启动完成,成功: {success_count}/{len(futures)}")
self._emit_event('streaming_started', {'active_devices': success_count})
else:
self.logger.error("没有设备数据流启动成功")
return self.is_running
except Exception as e:
self.logger.error(f"启动设备数据流失败: {e}")
return False
def stop_all_streaming(self) -> bool:
"""
停止所有设备的数据流
Returns:
bool: 停止是否成功
"""
with self.coordinator_lock:
if not self.is_running:
self.logger.warning("设备流未运行")
return True
try:
self.logger.info("停止所有设备数据流...")
# 并行停止所有设备流
futures = []
for device_name, device in self.devices.items():
if hasattr(device, 'stop_streaming'):
future = self.executor.submit(device.stop_streaming)
futures.append((device_name, future))
# 等待所有设备停止完成
for device_name, future in futures:
try:
future.result(timeout=5) # 5秒超时
self.logger.info(f"{device_name}数据流已停止")
except Exception as e:
self.logger.error(f"停止{device_name}数据流异常: {e}")
self.is_running = False
self.logger.info("所有设备数据流已停止")
self._emit_event('streaming_stopped', {})
return True
except Exception as e:
self.logger.error(f"停止设备数据流失败: {e}")
return False
def get_device_status(self, device_name: Optional[str] = None) -> Dict[str, Any]:
"""
获取设备状态
Args:
device_name: 设备名称None表示获取所有设备状态
Returns:
Dict[str, Any]: 设备状态信息
"""
if device_name:
if device_name in self.devices:
return self.devices[device_name].get_status()
else:
return {'error': f'设备 {device_name} 不存在'}
else:
# 获取所有设备状态
status = {
'coordinator': {
'is_initialized': self.is_initialized,
'is_running': self.is_running,
'device_count': len(self.devices),
'uptime': time.time() - self.stats['start_time'] if self.stats['start_time'] else 0
},
'devices': {}
}
for name, device in self.devices.items():
try:
status['devices'][name] = device.get_status()
except Exception as e:
status['devices'][name] = {'error': str(e)}
return status
def get_device(self, device_name: str) -> Optional[Any]:
"""
获取指定设备实例
Args:
device_name: 设备名称
Returns:
Optional[Any]: 设备实例不存在返回None
"""
return self.devices.get(device_name)
def get_device_managers(self) -> Dict[str, Any]:
"""
获取所有设备管理器实例
Returns:
Dict[str, Any]: 设备管理器字典
"""
return self.devices.copy()
def start_all_connection_monitor(self) -> bool:
"""
启动所有设备的连接监控
Returns:
bool: 启动是否成功
"""
with self.coordinator_lock:
if not self.is_initialized:
self.logger.error("设备协调器未初始化")
return False
try:
self.logger.info("启动所有设备连接监控...")
success_count = 0
for device_name, device in self.devices.items():
try:
# 对深度相机(femtobolt)和普通相机(camera)直接调用初始化和启动推流
2025-10-11 16:58:52 +08:00
if device_name in ['femtobolt', 'camera',"imu"]:
continue
if hasattr(device, '_start_connection_monitor'):
device._start_connection_monitor()
success_count += 1
self.logger.info(f"{device_name}设备连接监控已启动")
else:
self.logger.warning(f"{device_name}设备不支持连接监控")
except Exception as e:
self.logger.error(f"启动{device_name}设备失败: {e}")
self.logger.info(f"设备连接监控启动完成,成功: {success_count}/{len(self.devices)}")
return success_count > 0
except Exception as e:
self.logger.error(f"启动设备连接监控失败: {e}")
return False
def stop_all_connection_monitor(self) -> bool:
"""
停止所有设备的连接监控
Returns:
bool: 停止是否成功
"""
with self.coordinator_lock:
try:
self.logger.info("停止所有设备连接监控...")
success_count = 0
for device_name, device in self.devices.items():
try:
# 对深度相机(femtobolt)和普通相机(camera)直接调用停止推流
2025-10-11 16:58:52 +08:00
if device_name in ['femtobolt', 'camera',"imu"]:
self.logger.info(f"停止{device_name}设备推流")
# # 调用设备的cleanup方法清理资源,停止推流
# if hasattr(device, 'cleanup'):
# if device.cleanup():
# success_count += 1
# self.logger.info(f"{device_name}设备推流已停止")
# else:
# self.logger.warning(f"{device_name}设备推流停止失败")
# else:
# self.logger.warning(f"{device_name}设备不支持推流停止")
continue
if hasattr(device, '_stop_connection_monitor'):
device._stop_connection_monitor()
success_count += 1
self.logger.info(f"{device_name}设备连接监控已停止")
else:
self.logger.warning(f"{device_name}设备不支持连接监控")
except Exception as e:
self.logger.error(f"停止{device_name}设备失败: {e}")
self.logger.info(f"设备连接监控停止完成,成功: {success_count}/{len(self.devices)}")
return True
except Exception as e:
self.logger.error(f"停止设备连接监控失败: {e}")
return False
def restart_device(self, device_name: str) -> bool:
"""
2025-09-29 08:50:59 +08:00
彻底重启指定设备停止推流断开连接销毁设备实例重新创建实例初始化恢复推流
Args:
device_name: 设备名称
Returns:
bool: 重启是否成功
"""
if device_name not in self.devices:
self.logger.error(f"设备 {device_name} 不存在")
return False
2025-09-29 08:50:59 +08:00
restart_start = time.time()
device = self.devices[device_name]
was_streaming = False
try:
2025-09-29 08:50:59 +08:00
self.logger.info(f"开始彻底重启设备: {device_name}")
2025-09-29 08:50:59 +08:00
# 第一步:检查并停止数据流
stop_start = time.time()
if hasattr(device, 'is_streaming'):
was_streaming = device.is_streaming
2025-09-29 08:50:59 +08:00
if hasattr(device, 'stop_streaming') and was_streaming:
self.logger.info(f"正在停止 {device_name} 设备推流...")
try:
if not device.stop_streaming():
self.logger.warning(f"停止 {device_name} 推流失败,继续重启流程")
else:
self.logger.info(f"{device_name} 设备推流已停止")
except Exception as e:
self.logger.warning(f"停止 {device_name} 推流异常: {e},继续重启流程")
stop_time = (time.time() - stop_start) * 1000
# 第二步:断开连接并彻底清理资源
cleanup_start = time.time()
self.logger.info(f"正在彻底清理 {device_name} 设备...")
# 断开连接但暂时不广播状态变化,避免重启过程中的状态冲突
2025-09-29 08:50:59 +08:00
if hasattr(device, 'disconnect'):
try:
device.disconnect()
self.logger.info(f"{device_name} 设备连接已断开")
except Exception as e:
self.logger.warning(f"断开 {device_name} 连接异常: {e}")
# 静默设置设备状态为未连接,不触发状态变化通知
# 这样可以避免在重启过程中广播中间状态
if hasattr(device, 'is_connected'):
device.is_connected = False
self.logger.info(f"{device_name} 设备状态已静默更新为未连接(重启过程中)")
2025-09-29 08:50:59 +08:00
# 彻底清理资源
if hasattr(device, 'cleanup'):
2025-09-29 08:50:59 +08:00
try:
device.cleanup()
self.logger.info(f"{device_name} 设备资源已彻底清理")
except Exception as e:
self.logger.warning(f"清理 {device_name} 资源异常: {e}")
cleanup_time = (time.time() - cleanup_start) * 1000
# 第三步:彻底销毁设备实例
destroy_start = time.time()
self.logger.info(f"正在销毁 {device_name} 设备实例...")
# 从设备字典中移除
old_device = self.devices.pop(device_name, None)
if old_device:
# 强制删除引用,帮助垃圾回收
del old_device
self.logger.info(f"{device_name} 设备实例已销毁")
# 短暂等待,确保资源完全释放
time.sleep(0.2)
destroy_time = (time.time() - destroy_start) * 1000
# 第四步:重新创建设备实例
create_start = time.time()
self.logger.info(f"正在重新创建 {device_name} 设备实例...")
new_device = None
try:
# 根据设备类型重新创建实例
if device_name == 'camera':
try:
from .camera_manager import CameraManager
except ImportError:
from camera_manager import CameraManager
new_device = CameraManager(self.socketio, self.config_manager)
elif device_name == 'imu':
try:
from .imu_manager import IMUManager
except ImportError:
from imu_manager import IMUManager
new_device = IMUManager(self.socketio, self.config_manager)
elif device_name == 'pressure':
try:
from .pressure_manager import PressureManager
except ImportError:
from pressure_manager import PressureManager
new_device = PressureManager(self.socketio, self.config_manager)
elif device_name == 'femtobolt':
try:
from .femtobolt_manager import FemtoBoltManager
except ImportError:
from femtobolt_manager import FemtoBoltManager
new_device = FemtoBoltManager(self.socketio, self.config_manager)
else:
raise ValueError(f"未知的设备类型: {device_name}")
2025-09-29 08:50:59 +08:00
if new_device is None:
raise Exception("设备实例创建失败")
# 将新设备实例添加到设备字典
self.devices[device_name] = new_device
create_time = (time.time() - create_start) * 1000
self.logger.info(f"{device_name} 设备实例重新创建成功 (耗时: {create_time:.1f}ms)")
# 重新注册状态变化回调(如果有的话)
if hasattr(self, '_status_change_callback') and self._status_change_callback:
if hasattr(new_device, 'add_status_change_callback'):
new_device.add_status_change_callback(self._status_change_callback)
self.logger.info(f"{device_name} 设备状态变化回调已重新注册")
2025-09-29 08:50:59 +08:00
except Exception as e:
create_time = (time.time() - create_start) * 1000
self.logger.error(f"重新创建 {device_name} 设备实例失败: {e} (耗时: {create_time:.1f}ms)")
return False
2025-09-29 08:50:59 +08:00
# 第五步:初始化新设备实例
init_start = time.time()
self.logger.info(f"正在初始化新的 {device_name} 设备实例...")
if not new_device.initialize():
init_time = (time.time() - init_start) * 1000
self.logger.error(f"{device_name} 设备初始化失败 (耗时: {init_time:.1f}ms)")
# 初始化失败,从设备字典中移除
self.devices.pop(device_name, None)
return False
init_time = (time.time() - init_start) * 1000
self.logger.info(f"{device_name} 设备初始化成功 (耗时: {init_time:.1f}ms)")
# 设备初始化成功后,确保状态广播正确
# 此时设备应该已经通过initialize()方法中的set_connected(True)触发了状态变化通知
# 但为了确保状态一致性,我们再次确认状态
if hasattr(new_device, 'is_connected') and new_device.is_connected:
self.logger.info(f"{device_name} 设备重启后状态确认:已连接")
else:
self.logger.warning(f"{device_name} 设备重启后状态异常:未连接")
2025-09-29 08:50:59 +08:00
# 第六步:如果之前在推流,则启动推流
stream_time = 0
if was_streaming and hasattr(new_device, 'start_streaming'):
stream_start = time.time()
self.logger.info(f"正在启动 {device_name} 设备推流...")
try:
if not new_device.start_streaming():
stream_time = (time.time() - stream_start) * 1000
self.logger.error(f"启动 {device_name} 设备推流失败 (耗时: {stream_time:.1f}ms)")
return False
stream_time = (time.time() - stream_start) * 1000
self.logger.info(f"{device_name} 设备推流已启动 (耗时: {stream_time:.1f}ms)")
except Exception as e:
stream_time = (time.time() - stream_start) * 1000
self.logger.error(f"启动 {device_name} 推流异常: {e} (耗时: {stream_time:.1f}ms)")
return False
# 计算总耗时并记录
total_time = (time.time() - restart_start) * 1000
self.logger.info(f"{device_name} 设备彻底重启完成 - 停止推流: {stop_time:.1f}ms, 清理资源: {cleanup_time:.1f}ms, 销毁实例: {destroy_time:.1f}ms, 创建实例: {create_time:.1f}ms, 初始化: {init_time:.1f}ms, 启动推流: {stream_time:.1f}ms, 总耗时: {total_time:.1f}ms")
return True
except Exception as e:
2025-09-29 08:50:59 +08:00
total_time = (time.time() - restart_start) * 1000
error_msg = f"彻底重启设备 {device_name} 异常: {e} (耗时: {total_time:.1f}ms)"
self.logger.error(error_msg)
return False
def _start_monitor(self):
"""
启动监控线程
"""
if self.monitor_thread and self.monitor_thread.is_alive():
return
self.monitor_thread = threading.Thread(
target=self._monitor_worker,
name="DeviceMonitor",
daemon=True
)
self.monitor_thread.start()
self.logger.info("设备监控线程已启动")
def _monitor_worker(self):
"""
监控工作线程
"""
self.logger.info("设备监控线程开始运行")
while self.is_initialized:
try:
# 检查设备健康状态
for device_name, device in self.devices.items():
try:
status = device.get_status()
if not status.get('is_connected', False):
self.logger.warning(f"设备 {device_name} 连接丢失")
self.stats['device_errors'][device_name] += 1
# 尝试重连
if self.stats['device_errors'][device_name] <= 3:
self.logger.info(f"尝试重连设备: {device_name}")
if self.restart_device(device_name):
self.stats['device_errors'][device_name] = 0
except Exception as e:
self.logger.error(f"检查设备 {device_name} 状态异常: {e}")
# 发送状态更新
if self.is_running:
status = self.get_device_status()
self.socket_manager.emit_to_namespace(
'/coordinator', 'status_update', status
)
time.sleep(self.health_check_interval)
except Exception as e:
self.logger.error(f"监控线程异常: {e}")
time.sleep(1.0)
self.logger.info("设备监控线程结束")
def register_event_callback(self, event_name: str, callback: Callable):
"""
注册事件回调
Args:
event_name: 事件名称
callback: 回调函数
"""
self.event_callbacks[event_name].append(callback)
def _emit_event(self, event_name: str, data: Dict[str, Any]):
"""
触发事件
Args:
event_name: 事件名称
data: 事件数据
"""
# 调用注册的回调
for callback in self.event_callbacks[event_name]:
try:
callback(data)
except Exception as e:
self.logger.error(f"事件回调异常 {event_name}: {e}")
# 发送到Socket.IO
self.socket_manager.emit_to_namespace(
'/coordinator', event_name, data
)
def _cleanup_devices(self):
"""
清理所有设备
"""
for device_name, device in self.devices.items():
try:
if hasattr(device, 'cleanup'):
device.cleanup()
self.logger.info(f"设备 {device_name} 清理完成")
except Exception as e:
self.logger.error(f"清理设备 {device_name} 失败: {e}")
self.devices.clear()
def shutdown(self):
"""
关闭设备协调器
"""
with self.coordinator_lock:
try:
self.logger.info("关闭设备协调器...")
# 停止数据流
self.stop_all_streaming()
# 停止监控
self.is_initialized = False
if self.monitor_thread and self.monitor_thread.is_alive():
self.monitor_thread.join(timeout=5.0)
# 清理设备
self._cleanup_devices()
# 关闭线程池
self.executor.shutdown(wait=True)
# 清理Socket管理器
self.socket_manager.cleanup()
self.logger.info("设备协调器已关闭")
except Exception as e:
self.logger.error(f"关闭设备协调器失败: {e}")
def __enter__(self):
"""上下文管理器入口"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""上下文管理器出口"""
2025-09-29 08:50:59 +08:00
self.shutdown()
def test_restart_device(device_name=None):
"""
测试设备重启功能
该测试方法演示如何使用restart_device方法进行设备的彻底重启
包括模拟设备初始化推流重启和状态验证的完整流程
Args:
device_name (str, optional): 指定要测试的设备名称如果为None则自动选择第一个可用设备
可选值: 'camera', 'imu', 'pressure', 'femtobolt'
"""
import time
import threading
from unittest.mock import Mock
print("=" * 60)
print("设备协调器重启功能测试")
print("=" * 60)
# 创建模拟的SocketIO和配置管理器
mock_socketio = Mock()
mock_config_manager = Mock()
# 模拟配置数据
mock_config_manager.get_device_config.return_value = {
'camera': {'enabled': True, 'device_id': 0, 'fps': 30},
'imu': {'enabled': True, 'device_type': 'mock'},
'pressure': {'enabled': True, 'device_type': 'mock'},
'femtobolt': {'enabled': False}
}
try:
# 创建设备协调器实例
print("1. 创建设备协调器...")
coordinator = DeviceCoordinator(mock_socketio, mock_config_manager)
# 初始化设备协调器
print("2. 初始化设备协调器...")
if not coordinator.initialize():
print("❌ 设备协调器初始化失败")
return False
print("✅ 设备协调器初始化成功")
print(f" 已初始化设备: {list(coordinator.devices.keys())}")
# 等待设备稳定
time.sleep(1)
# 选择要测试的设备
available_devices = list(coordinator.devices.keys())
if not available_devices:
print("❌ 没有可用的设备进行测试")
return False
# 根据参数选择测试设备
if device_name:
if device_name in available_devices:
test_device = device_name
print(f"3. 使用指定的测试设备: {test_device}")
else:
print(f"❌ 指定的设备 '{device_name}' 不存在")
print(f" 可用设备: {available_devices}")
return False
else:
test_device = available_devices[0] # 选择第一个设备进行测试
print(f"3. 自动选择测试设备: {test_device}")
print(f" 可用设备: {available_devices}")
# 获取设备初始状态
device = coordinator.devices[test_device]
initial_streaming = getattr(device, 'is_streaming', False)
initial_connected = getattr(device, 'is_connected', False)
print(f" 设备初始状态 - 连接: {initial_connected}, 推流: {initial_streaming}")
# 如果设备未推流,先启动推流
if hasattr(device, 'start_streaming') and not initial_streaming:
print("4. 启动设备推流...")
if device.start_streaming():
print("✅ 设备推流启动成功")
time.sleep(0.5) # 等待推流稳定
else:
print("⚠️ 设备推流启动失败,继续测试")
# 记录重启前的设备实例ID
old_device_id = id(device)
print(f" 重启前设备实例ID: {old_device_id}")
# 执行设备重启
print("5. 执行设备重启...")
restart_start = time.time()
restart_success = coordinator.restart_device(test_device)
restart_time = (time.time() - restart_start) * 1000
if restart_success:
print(f"✅ 设备重启成功 (总耗时: {restart_time:.1f}ms)")
# 验证设备实例是否已更换
new_device = coordinator.devices.get(test_device)
if new_device:
new_device_id = id(new_device)
print(f" 重启后设备实例ID: {new_device_id}")
if new_device_id != old_device_id:
print("✅ 设备实例已成功更换")
else:
print("❌ 设备实例未更换,可能重启不彻底")
# 检查设备状态
new_connected = getattr(new_device, 'is_connected', False)
new_streaming = getattr(new_device, 'is_streaming', False)
print(f" 重启后设备状态 - 连接: {new_connected}, 推流: {new_streaming}")
# 验证推流状态恢复
if initial_streaming and new_streaming:
print("✅ 推流状态已正确恢复")
elif not initial_streaming and not new_streaming:
print("✅ 推流状态保持一致")
else:
print("⚠️ 推流状态与预期不符")
else:
print("❌ 重启后设备实例丢失")
else:
print(f"❌ 设备重启失败 (耗时: {restart_time:.1f}ms)")
print("8. 清理资源...")
coordinator.shutdown()
print("✅ 资源清理完成")
print("=" * 60)
print("测试完成")
print("=" * 60)
return restart_success
except Exception as e:
print(f"❌ 测试过程中发生异常: {e}")
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
"""
直接运行此文件时执行设备重启测试
"""
print("启动设备协调器重启功能测试...")
# 检查命令行参数
device_name = None
try:
# 设置日志级别
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 执行测试
# 可选值: 'camera', 'imu', 'pressure', 'femtobolt'
success = test_restart_device('pressure')
if success:
print("\n🎉 所有测试通过!")
else:
print("\n❌ 测试失败,请检查日志")
except KeyboardInterrupt:
print("\n⚠️ 测试被用户中断")
except Exception as e:
print(f"\n❌ 测试启动失败: {e}")
import traceback
traceback.print_exc()