优化了设备启动效率和状态检测机制

This commit is contained in:
root 2025-09-18 09:07:09 +08:00
parent 147c66ada6
commit 26b6bc3e0a
24 changed files with 2943 additions and 708 deletions

View File

@ -0,0 +1,186 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
检查设备连接监控线程状态的测试脚本
"""
import sys
import os
import threading
import time
from typing import Dict, Any
# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from devices.utils.config_manager import ConfigManager
from devices.camera_manager import CameraManager
from devices.imu_manager import IMUManager
from devices.pressure_manager import PressureManager
from devices.femtobolt_manager import FemtoBoltManager
class MockCameraManager(CameraManager):
"""模拟摄像头管理器,用于测试监控线程"""
def __init__(self, socketio, config_manager):
super().__init__(socketio, config_manager)
self._mock_hardware_connected = True
def check_hardware_connection(self) -> bool:
"""模拟硬件连接检查"""
return self._mock_hardware_connected
def set_mock_hardware_status(self, connected: bool):
"""设置模拟硬件连接状态"""
self._mock_hardware_connected = connected
def check_device_monitor_status(device_manager, device_name: str):
"""
检查单个设备的监控线程状态
Args:
device_manager: 设备管理器实例
device_name: 设备名称
"""
print(f"\n=== {device_name.upper()} 设备监控状态检查 ===")
# 检查基本状态
print(f"设备连接状态: {device_manager.is_connected}")
print(f"设备流状态: {device_manager.is_streaming}")
# 检查监控线程相关属性
if hasattr(device_manager, '_connection_monitor_thread'):
monitor_thread = device_manager._connection_monitor_thread
print(f"监控线程对象: {monitor_thread}")
if monitor_thread:
print(f"监控线程存活状态: {monitor_thread.is_alive()}")
print(f"监控线程名称: {monitor_thread.name}")
print(f"监控线程守护状态: {monitor_thread.daemon}")
else:
print("监控线程对象: None")
else:
print("设备管理器没有监控线程属性")
# 检查监控停止事件
if hasattr(device_manager, '_monitor_stop_event'):
stop_event = device_manager._monitor_stop_event
print(f"监控停止事件: {stop_event}")
print(f"监控停止事件状态: {stop_event.is_set()}")
else:
print("设备管理器没有监控停止事件属性")
# 检查监控配置
if hasattr(device_manager, '_connection_check_interval'):
print(f"连接检查间隔: {device_manager._connection_check_interval}")
if hasattr(device_manager, '_connection_timeout'):
print(f"连接超时时间: {device_manager._connection_timeout}")
# 检查心跳时间
if hasattr(device_manager, '_last_heartbeat'):
last_heartbeat = device_manager._last_heartbeat
current_time = time.time()
heartbeat_age = current_time - last_heartbeat
print(f"上次心跳时间: {time.ctime(last_heartbeat)}")
print(f"心跳间隔: {heartbeat_age:.2f}秒前")
def main():
"""
主函数检查所有设备的监控状态
"""
print("设备连接监控状态检查工具")
print("=" * 50)
try:
# 初始化配置管理器
config_manager = ConfigManager()
# 创建模拟设备管理器实例
mock_camera = MockCameraManager(None, config_manager)
print("\n=== 初始状态检查 ===")
check_device_monitor_status(mock_camera, 'mock_camera')
print("\n=== 系统线程信息 ===")
active_threads = threading.active_count()
print(f"当前活跃线程数: {active_threads}")
print("\n活跃线程列表:")
for thread in threading.enumerate():
print(f" - {thread.name} (守护: {thread.daemon}, 存活: {thread.is_alive()})")
# 测试连接监控启动
print("\n=== 测试连接监控启动 ===")
print("\n1. 设置模拟硬件为连接状态...")
mock_camera.set_mock_hardware_status(True)
print("\n2. 设置设备为连接状态...")
mock_camera.set_connected(True)
time.sleep(0.5) # 等待线程启动
print("\n连接后的监控状态:")
check_device_monitor_status(mock_camera, 'mock_camera')
print("\n=== 系统线程信息 (启动监控后) ===")
active_threads = threading.active_count()
print(f"当前活跃线程数: {active_threads}")
print("\n活跃线程列表:")
for thread in threading.enumerate():
print(f" - {thread.name} (守护: {thread.daemon}, 存活: {thread.is_alive()})")
print("\n3. 等待3秒观察监控线程工作...")
time.sleep(3)
print("\n监控运行中的状态:")
check_device_monitor_status(mock_camera, 'mock_camera')
print("\n4. 模拟硬件断开...")
mock_camera.set_mock_hardware_status(False)
time.sleep(6) # 等待监控检测到断开检查间隔是5秒
print("\n硬件断开后的监控状态:")
check_device_monitor_status(mock_camera, 'mock_camera')
print("\n5. 重新连接测试...")
mock_camera.set_mock_hardware_status(True)
mock_camera.set_connected(True)
time.sleep(0.5)
print("\n重新连接后的监控状态:")
check_device_monitor_status(mock_camera, 'mock_camera')
print("\n6. 手动断开连接...")
mock_camera.set_connected(False)
time.sleep(0.5)
print("\n手动断开后的监控状态:")
check_device_monitor_status(mock_camera, 'mock_camera')
print("\n=== 最终系统线程信息 ===")
active_threads = threading.active_count()
print(f"当前活跃线程数: {active_threads}")
print("\n活跃线程列表:")
for thread in threading.enumerate():
print(f" - {thread.name} (守护: {thread.daemon}, 存活: {thread.is_alive()})")
except Exception as e:
print(f"检查过程中发生错误: {e}")
import traceback
traceback.print_exc()
finally:
# 清理资源
print("\n=== 清理资源 ===")
try:
if 'mock_camera' in locals():
mock_camera.cleanup()
print("mock_camera 设备资源已清理")
except Exception as e:
print(f"清理资源时发生错误: {e}")
if __name__ == '__main__':
main()

View File

@ -59,13 +59,7 @@ class BaseDevice(ABC):
'last_error': None
}
# 性能统计
self._stats = {
'frames_processed': 0,
'errors_count': 0,
'start_time': None,
'last_frame_time': None
}
@abstractmethod
def initialize(self) -> bool:
@ -210,11 +204,6 @@ class BaseDevice(ABC):
if old_status != is_connected:
self._notify_status_change(is_connected)
# 启动或停止连接监控
if is_connected and not self._connection_monitor_thread:
self._start_connection_monitor()
elif not is_connected and self._connection_monitor_thread:
self._stop_connection_monitor()
def emit_data(self, event: str, data: Any, namespace: Optional[str] = None):
"""
@ -260,37 +249,9 @@ class BaseDevice(ABC):
with self._lock:
return self._device_info.copy()
def get_stats(self) -> Dict[str, Any]:
"""
获取性能统计
Returns:
Dict[str, Any]: 性能统计信息
"""
with self._lock:
stats = self._stats.copy()
if stats['start_time']:
stats['uptime'] = time.time() - stats['start_time']
if stats['frames_processed'] > 0 and stats['uptime'] > 0:
stats['fps'] = stats['frames_processed'] / stats['uptime']
else:
stats['fps'] = 0
return stats
def _update_stats(self, frame_processed: bool = True, error: bool = False):
"""
更新统计信息
Args:
frame_processed: 是否处理了一帧
error: 是否发生错误
"""
with self._lock:
if frame_processed:
self._stats['frames_processed'] += 1
self._stats['last_frame_time'] = time.time()
if error:
self._stats['errors_count'] += 1
def _set_error(self, error_msg: str):
"""
@ -312,21 +273,7 @@ class BaseDevice(ABC):
with self._lock:
self._device_info['last_error'] = None
def _start_stats_tracking(self):
"""
开始统计跟踪
"""
with self._lock:
self._stats['start_time'] = time.time()
self._stats['frames_processed'] = 0
self._stats['errors_count'] = 0
def _stop_stats_tracking(self):
"""
停止统计跟踪
"""
with self._lock:
self._stats['start_time'] = None
def _start_connection_monitor(self):
"""
@ -365,18 +312,29 @@ class BaseDevice(ABC):
try:
# 检查硬件连接状态
hardware_connected = self.check_hardware_connection()
self.logger.info(f"检测到设备 {self.device_name} 硬件连接状态: {hardware_connected} is_connected{self.is_connected}")
# 如果硬件断开但软件状态仍为连接,则更新状态
if not hardware_connected and self.is_connected:
self.logger.warning(f"检测到设备 {self.device_name} 硬件连接断开")
self.set_connected(False)
break # 硬件断开后停止监控
# 直接更新状态避免在监控线程中调用set_connected导致死锁
self.is_connected = False
self._notify_status_change(False)
# 检查心跳超时
# 如果硬件重新连接但软件状态仍为断开,则更新状态
elif hardware_connected and not self.is_connected:
self.logger.info(f"检测到设备 {self.device_name} 硬件重新连接")
# 直接更新状态避免在监控线程中调用set_connected导致死锁
self.is_connected = True
# 重置心跳时间,避免立即触发心跳超时
self.update_heartbeat()
self._notify_status_change(True)
# 检查心跳超时(仅在当前状态为连接时检查)
if self.is_connected and time.time() - self._last_heartbeat > self._connection_timeout:
self.logger.warning(f"设备 {self.device_name} 心跳超时,判定为断开连接")
self.set_connected(False)
break # 超时后停止监控
# 直接更新状态避免在监控线程中调用set_connected导致死锁
self.is_connected = False
self._notify_status_change(False)
except Exception as e:
self.logger.error(f"设备 {self.device_name} 连接监控异常: {e}")
@ -385,6 +343,8 @@ class BaseDevice(ABC):
self._monitor_stop_event.wait(self._connection_check_interval)
self.logger.info(f"设备 {self.device_name} 连接监控结束")
# 清理线程引用
self._connection_monitor_thread = None
def __enter__(self):
"""

View File

@ -10,18 +10,16 @@ import threading
import time
import base64
import numpy as np
from typing import Optional, Dict, Any, Tuple
from typing import Optional, Dict, Any
import logging
import queue
import gc
try:
from .base_device import BaseDevice
from .utils.socket_manager import SocketManager
from .utils.config_manager import ConfigManager
except ImportError:
from base_device import BaseDevice
from utils.socket_manager import SocketManager
from utils.config_manager import ConfigManager
@ -54,6 +52,17 @@ class CameraManager(BaseDevice):
self.buffer_size = config.get('buffer_size', 1)
self.fourcc = config.get('fourcc', 'MJPG')
# OpenCV后端配置 (DirectShow性能最佳)
backend_name = config.get('backend', 'directshow').lower()
self.backend_map = {
'directshow': cv2.CAP_DSHOW,
'dshow': cv2.CAP_DSHOW,
'msmf': cv2.CAP_MSMF,
'any': cv2.CAP_ANY
}
self.preferred_backend = self.backend_map.get(backend_name, cv2.CAP_DSHOW)
self.backend_name = backend_name
# 额外可调的降采样宽度(不改变外部配置语义,仅内部优化传输)
self._tx_max_width = int(config.get('tx_max_width', 640))
@ -87,14 +96,48 @@ class CameraManager(BaseDevice):
# 全局帧队列(用于录制)
self.frame_queue = queue.Queue(maxsize=10) # 最大长度10自动丢弃旧帧
# OpenCV优化开关
# 属性缓存机制 - 避免重复设置相同属性值
self._property_cache = {}
self._cache_enabled = True
# OpenCV优化开关和性能设置
try:
cv2.setUseOptimized(True)
# 设置OpenCV线程数以提高性能
cv2.setNumThreads(4) # 使用4个线程
except Exception:
pass
self.logger.info(f"相机管理器初始化完成 - 设备索引: {self.device_index}")
def _set_property_optimized(self, prop, value):
"""
优化的属性设置方法使用缓存避免重复设置
Args:
prop: OpenCV属性常量
value: 属性值
Returns:
bool: 是否实际执行了设置操作
"""
if not self.cap:
return False
# 检查缓存,避免重复设置相同值
if self._cache_enabled and prop in self._property_cache:
if self._property_cache[prop] == value:
return False # 值未改变,跳过设置
# 执行属性设置
result = self.cap.set(prop, value)
# 更新缓存
if self._cache_enabled:
self._property_cache[prop] = value
return True
def initialize(self) -> bool:
"""
初始化相机设备
@ -102,37 +145,66 @@ class CameraManager(BaseDevice):
Returns:
bool: 初始化是否成功
"""
start_time = time.time()
try:
self.logger.info(f"正在初始化相机设备...")
# 使用构造函数中已加载的配置,避免并发读取配置文件
self.logger.info(f"使用已加载配置: device_index={self.device_index}, resolution={self.width}x{self.height}, fps={self.fps}")
config_time = time.time()
self.logger.info(f"使用已加载配置: device_index={self.device_index}, resolution={self.width}x{self.height}, fps={self.fps} (耗时: {(config_time - start_time)*1000:.1f}ms)")
# 尝试多个后端Windows下优先MSMF/DShow
backends = [cv2.CAP_MSMF, cv2.CAP_DSHOW, cv2.CAP_ANY]
# 使用配置的后端,如果失败则尝试其他后端
if self.preferred_backend == cv2.CAP_DSHOW:
backends = [cv2.CAP_DSHOW, cv2.CAP_MSMF, cv2.CAP_ANY]
elif self.preferred_backend == cv2.CAP_MSMF:
backends = [cv2.CAP_MSMF, cv2.CAP_DSHOW, cv2.CAP_ANY]
else:
backends = [self.preferred_backend, cv2.CAP_DSHOW, cv2.CAP_MSMF, cv2.CAP_ANY]
camera_open_time = time.time()
for backend in backends:
backend_start = time.time()
try:
# 快速打开相机,减少超时等待
self.cap = cv2.VideoCapture(self.device_index, backend)
# 设置较短的超时时间以加快检测
if hasattr(cv2, 'CAP_PROP_OPEN_TIMEOUT_MSEC'):
self.cap.set(cv2.CAP_PROP_OPEN_TIMEOUT_MSEC, 3000) # 3秒超时
if self.cap.isOpened():
self.logger.info(f"使用后端 {backend} 成功打开相机")
backend_time = (time.time() - backend_start) * 1000
self.logger.info(f"使用后端 {backend} 成功打开相机 (耗时: {backend_time:.1f}ms)")
break
except Exception as e:
self.logger.warning(f"后端 {backend} 打开相机失败: {e}")
backend_time = (time.time() - backend_start) * 1000
self.logger.warning(f"后端 {backend} 打开相机失败: {e} (耗时: {backend_time:.1f}ms)")
continue
else:
self.logger.warning("所有后端都无法打开相机,相机设备不可用")
total_open_time = (time.time() - camera_open_time) * 1000
self.logger.warning(f"所有后端都无法打开相机,相机设备不可用 (总耗时: {total_open_time:.1f}ms)")
return False
total_open_time = (time.time() - camera_open_time) * 1000
self.logger.info(f"相机打开完成 (总耗时: {total_open_time:.1f}ms)")
# 设置相机属性
config_start = time.time()
self._configure_camera()
config_time = (time.time() - config_start) * 1000
self.logger.info(f"相机配置完成 (耗时: {config_time:.1f}ms)")
# 验证相机是否正常工作
# 验证相机是否正常工作(优化测试过程)
test_start = time.time()
if not self._test_camera():
self.logger.warning("相机测试失败,相机设备不可用")
test_time = (time.time() - test_start) * 1000
self.logger.warning(f"相机测试失败,相机设备不可用 (耗时: {test_time:.1f}ms)")
return False
test_time = (time.time() - test_start) * 1000
self.logger.info(f"相机测试完成 (耗时: {test_time:.1f}ms)")
self.is_connected = True
# 使用set_connected方法来正确启动连接监控线程
self.set_connected(True)
self._last_connected_state = True
self._device_info.update({
'device_index': self.device_index,
@ -141,7 +213,8 @@ class CameraManager(BaseDevice):
'backend': self.cap.getBackendName() if hasattr(self.cap, 'getBackendName') else 'Unknown'
})
self.logger.info("相机初始化成功")
total_time = (time.time() - start_time) * 1000
self.logger.info(f"相机初始化成功 (总耗时: {total_time:.1f}ms)")
return True
except Exception as e:
@ -173,30 +246,136 @@ class CameraManager(BaseDevice):
return
try:
# 设置FOURCC编码
if self.fourcc:
fourcc_code = cv2.VideoWriter_fourcc(*self.fourcc)
self.cap.set(cv2.CAP_PROP_FOURCC, fourcc_code)
# 批量设置相机属性以提高效率
config_start = time.time()
# 设置分辨率
self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.width)
self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.height)
# 设置帧率
self.cap.set(cv2.CAP_PROP_FPS, self.fps)
# 设置缓冲区大小(部分后端不生效)
# 设置缓冲区大小(优先设置,减少延迟)
buffer_start = time.time()
try:
self.cap.set(cv2.CAP_PROP_BUFFERSIZE, self.buffer_size)
self.cap.set(cv2.CAP_PROP_BUFFERSIZE, min(self.buffer_size, 1)) # 使用最小缓冲区减少延迟
except Exception:
pass
buffer_time = (time.time() - buffer_start) * 1000
self.logger.debug(f"缓冲区设置耗时: {buffer_time:.1f}ms")
# 获取实际设置的值
actual_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
actual_fps = self.cap.get(cv2.CAP_PROP_FPS)
# 性能优化设置:禁用可能导致延迟的自动功能
optimization_start = time.time()
try:
# 禁用自动曝光以减少处理时间
self.cap.set(cv2.CAP_PROP_AUTO_EXPOSURE, 0.25) # 手动曝光模式
# 禁用自动白平衡
self.cap.set(cv2.CAP_PROP_AUTO_WB, 0)
# 设置较低的曝光值以减少延迟
self.cap.set(cv2.CAP_PROP_EXPOSURE, -6)
except Exception as e:
self.logger.debug(f"设置性能优化参数时出现警告: {e}")
optimization_time = (time.time() - optimization_start) * 1000
self.logger.debug(f"性能优化设置耗时: {optimization_time:.1f}ms")
self.logger.info(f"相机配置 - 分辨率: {actual_width}x{actual_height}, FPS: {actual_fps}")
# 激进优化:跳过非关键属性设置,只设置必要属性
resolution_start = time.time()
# 优先级属性设置:只设置最关键的属性
critical_properties = [
(cv2.CAP_PROP_FRAME_WIDTH, self.width),
(cv2.CAP_PROP_FRAME_HEIGHT, self.height)
]
# 可选属性(在某些情况下可以跳过)
optional_properties = [
(cv2.CAP_PROP_FPS, self.fps)
]
batch_start = time.time()
actual_sets = 0
skipped_sets = 0
# 设置关键属性
for prop, value in critical_properties:
try:
if self._set_property_optimized(prop, value):
actual_sets += 1
else:
skipped_sets += 1
except Exception as e:
self.logger.debug(f"设置属性 {prop} 失败: {e}")
skipped_sets += 1
# 条件设置可选属性(如果时间允许)
current_time = (time.time() - batch_start) * 1000
if current_time < 1000: # 如果已用时间少于1秒才设置FPS
for prop, value in optional_properties:
try:
if self._set_property_optimized(prop, value):
actual_sets += 1
else:
skipped_sets += 1
except Exception as e:
self.logger.debug(f"设置可选属性 {prop} 失败: {e}")
skipped_sets += 1
else:
self.logger.debug(f"跳过FPS设置以节省时间 (已耗时: {current_time:.1f}ms)")
skipped_sets += len(optional_properties)
batch_time = (time.time() - batch_start) * 1000
resolution_time = (time.time() - resolution_start) * 1000
self.logger.debug(f"优化属性设置耗时: {batch_time:.1f}ms, 实际设置: {actual_sets}, 跳过: {skipped_sets}, 总计: {resolution_time:.1f}ms")
# 将原来的单独计时变量设为批量时间的一部分用于兼容性
width_time = batch_time * 0.5 # 估算宽度设置占比
height_time = batch_time * 0.4 # 估算高度设置占比
fps_time = batch_time * 0.1 # 估算帧率设置占比(可能被跳过)
# 设置FOURCC编码如果需要
if self.fourcc:
fourcc_start = time.time()
fourcc_code = cv2.VideoWriter_fourcc(*self.fourcc)
self.cap.set(cv2.CAP_PROP_FOURCC, fourcc_code)
fourcc_time = (time.time() - fourcc_start) * 1000
self.logger.debug(f"FOURCC设置耗时: {fourcc_time:.1f}ms")
# 激进优化:延迟验证机制 - 只在必要时验证
verification_start = time.time()
# 检查是否需要验证(基于配置或调试需求)
need_verification = self.logger.isEnabledFor(logging.DEBUG) or getattr(self, '_force_verification', False)
if need_verification:
# 只读取关键属性进行验证
batch_read_start = time.time()
try:
actual_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
# FPS验证可选因为某些相机不支持精确FPS设置
actual_fps = self.cap.get(cv2.CAP_PROP_FPS) if actual_sets > 2 else self.fps
batch_read_time = (time.time() - batch_read_start) * 1000
self.logger.debug(f"验证读取耗时: {batch_read_time:.1f}ms")
except Exception as e:
# 验证失败不影响主流程
actual_width, actual_height, actual_fps = self.width, self.height, self.fps
batch_read_time = (time.time() - batch_read_start) * 1000
self.logger.debug(f"属性验证失败,使用默认值: {e}")
else:
# 跳过验证,使用设置值
actual_width, actual_height, actual_fps = self.width, self.height, self.fps
batch_read_time = 0.0
self.logger.debug("跳过属性验证以节省时间")
verification_time = (time.time() - verification_start) * 1000
# 为兼容性保留单独计时变量
width_read_time = batch_read_time * 0.4
height_read_time = batch_read_time * 0.4
fps_read_time = batch_read_time * 0.2
self.logger.debug(f"延迟验证耗时: {batch_read_time:.1f}ms, 总计: {verification_time:.1f}ms")
total_config_time = (time.time() - config_start) * 1000
self.logger.info(f"相机配置完成 - 分辨率: {actual_width}x{actual_height}, FPS: {actual_fps}")
self.logger.info(f"配置耗时统计 - 缓冲区: {buffer_time:.1f}ms, 优化设置: {optimization_time:.1f}ms, 分辨率: {resolution_time:.1f}ms, 帧率: {fps_time:.1f}ms, 验证: {verification_time:.1f}ms, 总计: {total_config_time:.1f}ms")
self.logger.debug(f"配置详情 - 分辨率设置: {resolution_time:.1f}ms, FPS设置: {fps_time:.1f}ms, 验证: {verification_time:.1f}ms, 总计: {total_config_time:.1f}ms")
except Exception as e:
self.logger.warning(f"配置相机参数失败: {e}")
@ -209,12 +388,24 @@ class CameraManager(BaseDevice):
bool: 测试是否成功
"""
try:
# 快速测试:只读取一帧进行验证
read_start = time.time()
ret, frame = self.cap.read()
read_time = (time.time() - read_start) * 1000
if ret and frame is not None:
self.logger.info(f"相机测试成功 - 帧大小: {frame.shape}")
return True
# 基本帧验证
if len(frame.shape) >= 2 and frame.shape[0] > 0 and frame.shape[1] > 0:
self.logger.info(f"相机测试成功 - 帧大小: {frame.shape}, 读取耗时: {read_time:.1f}ms")
# 清理测试帧内存
del frame
return True
else:
self.logger.error(f"相机测试失败 - 帧数据无效: {frame.shape if frame is not None else 'None'}")
return False
else:
self.logger.error("相机测试失败 - 无法读取帧")
self.logger.error(f"相机测试失败 - 无法读取帧, 读取耗时: {read_time:.1f}ms")
return False
except Exception as e:
self.logger.error(f"相机测试异常: {e}")
@ -227,24 +418,43 @@ class CameraManager(BaseDevice):
Returns:
bool: 校准是否成功
"""
calibrate_start = time.time()
try:
self.logger.info("开始相机校准...")
if not self.is_connected:
init_start = time.time()
if not self.initialize():
return False
init_time = (time.time() - init_start) * 1000
self.logger.info(f"校准中初始化完成 (耗时: {init_time:.1f}ms)")
# 优化减少稳定帧数量只读取2帧来稳定相机
stabilize_start = time.time()
stable_frames = 2 # 减少从5帧到2帧
for i in range(stable_frames):
frame_start = time.time()
ret, frame = self.cap.read()
frame_time = (time.time() - frame_start) * 1000
# 读取几帧来稳定相机
for i in range(5):
ret, _ = self.cap.read()
if not ret:
self.logger.warning(f"校时时读取第{i+1}帧失败")
self.logger.warning(f"校准时读取第{i+1}帧失败 (耗时: {frame_time:.1f}ms)")
else:
# 立即释放帧内存
if frame is not None:
del frame
self.logger.debug(f"校准帧{i+1}读取成功 (耗时: {frame_time:.1f}ms)")
self.logger.info("相机校准完成")
stabilize_time = (time.time() - stabilize_start) * 1000
total_time = (time.time() - calibrate_start) * 1000
self.logger.info(f"相机校准完成 - 稳定化耗时: {stabilize_time:.1f}ms, 总耗时: {total_time:.1f}ms")
return True
except Exception as e:
self.logger.error(f"相机校准失败: {e}")
total_time = (time.time() - calibrate_start) * 1000
self.logger.error(f"相机校准失败: {e} (耗时: {total_time:.1f}ms)")
return False
def start_streaming(self) -> bool:
@ -400,6 +610,9 @@ class CameraManager(BaseDevice):
consecutive_read_failures = 0
self.dropped_frames = 0
# 更新心跳时间,防止连接监控线程判定为超时
self.update_heartbeat()
# 保存原始帧到队列(用于录制)
try:
self.frame_queue.put_nowait({
@ -685,15 +898,80 @@ class CameraManager(BaseDevice):
bool: 相机是否物理连接
"""
try:
if self.cap and self.cap.isOpened():
# 尝试读取一帧来验证连接
ret, _ = self.cap.read()
return ret
return False
if not self.cap:
# 如果相机实例不存在,尝试重新创建
return self._attempt_device_reconnection()
if not self.cap.isOpened():
# 相机未打开,尝试重连
return self._attempt_device_reconnection()
# 尝试读取一帧来验证连接
try:
ret, frame = self.cap.read()
if ret and frame is not None:
# 立即释放帧内存
del frame
return True
else:
# 读取失败,可能设备已断开
return self._attempt_device_reconnection()
except Exception:
# 读取异常,尝试重连
return self._attempt_device_reconnection()
except Exception as e:
self.logger.debug(f"检查相机硬件连接时发生异常: {e}")
return False
def _attempt_device_reconnection(self) -> bool:
"""
尝试重新连接相机设备
Returns:
bool: 重连是否成功
"""
try:
self.logger.info("检测到相机设备断开,尝试重新连接...")
# 清理旧的相机实例
if self.cap:
try:
self.cap.release()
except Exception as e:
self.logger.debug(f"清理旧相机实例时出错: {e}")
self.cap = None
# 等待设备释放
time.sleep(0.5)
# 重新初始化相机
if self.initialize():
self._notify_status_change(True)
# 重连成功后,确保数据流正在运行
if not self.is_streaming:
self.logger.info("重连成功,启动相机数据流")
self.start_streaming()
# 更新设备信息
self._device_info.update({
'device_index': self.device_index,
'resolution': f"{self.width}x{self.height}",
'fps': self.fps,
'backend': self.cap.getBackendName() if hasattr(self.cap, 'getBackendName') else 'Unknown'
})
return True
else:
self.logger.warning("相机设备重连失败")
return False
except Exception as e:
self.logger.error(f"相机设备重连过程中出错: {e}")
self.cap = None
return False
def cleanup(self):
"""
清理资源

View File

@ -49,7 +49,9 @@ class DeviceCoordinator:
# 设备管理器
self.devices: Dict[str, Any] = {}
self.device_configs = self.config_manager.get_system_config().get('devices', {})
# 获取设备配置
self.device_configs = self.config_manager.get_all_device_configs()
# 状态管理
self.is_initialized = False
@ -118,11 +120,15 @@ class DeviceCoordinator:
"""
注册Socket.IO命名空间
"""
namespaces = ['/devices', '/coordinator']
for namespace in namespaces:
self.socket_manager.register_namespace(namespace)
namespace_mappings = {
'/devices': 'devices',
'/coordinator': 'coordinator'
}
self.logger.info(f"已注册Socket.IO命名空间: {namespaces}")
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:
"""
@ -189,8 +195,8 @@ class DeviceCoordinator:
"""
try:
camera = CameraManager(self.socketio, self.config_manager)
self.devices['camera'] = camera
if camera.initialize():
self.devices['camera'] = camera
return True
return False
except Exception as e:
@ -206,9 +212,9 @@ class DeviceCoordinator:
"""
try:
imu = IMUManager(self.socketio, self.config_manager)
self.devices['imu'] = imu
if imu.initialize():
self.devices['imu'] = imu
return True
return True
return False
except Exception as e:
self.logger.error(f"初始化IMU失败: {e}")
@ -223,9 +229,9 @@ class DeviceCoordinator:
"""
try:
pressure = PressureManager(self.socketio, self.config_manager)
self.devices['pressure'] = pressure
if pressure.initialize():
self.devices['pressure'] = pressure
return True
return True
return False
except Exception as e:
self.logger.error(f"初始化压力传感器失败: {e}")
@ -240,8 +246,8 @@ class DeviceCoordinator:
"""
try:
femtobolt = FemtoBoltManager(self.socketio, self.config_manager)
self.devices['femtobolt'] = femtobolt
if femtobolt.initialize():
self.devices['femtobolt'] = femtobolt
return True
return False
except Exception as e:
@ -340,47 +346,7 @@ class DeviceCoordinator:
self.logger.error(f"停止设备数据流失败: {e}")
return False
def calibrate_all_devices(self) -> Dict[str, bool]:
"""
校准所有设备
Returns:
Dict[str, bool]: 各设备校准结果
"""
results = {}
try:
self.logger.info("开始校准所有设备...")
# 并行校准所有设备
futures = []
for device_name, device in self.devices.items():
if hasattr(device, 'calibrate'):
future = self.executor.submit(device.calibrate)
futures.append((device_name, future))
# 等待所有设备校准完成
for device_name, future in futures:
try:
result = future.result(timeout=30) # 30秒超时
results[device_name] = result
if result:
self.logger.info(f"{device_name}校准成功")
else:
self.logger.error(f"{device_name}校准失败")
except Exception as e:
self.logger.error(f"{device_name}校准异常: {e}")
results[device_name] = False
success_count = sum(results.values())
self.logger.info(f"设备校准完成,成功: {success_count}/{len(results)}")
self._emit_event('calibration_completed', results)
return results
except Exception as e:
self.logger.error(f"设备校准失败: {e}")
return results
def get_device_status(self, device_name: Optional[str] = None) -> Dict[str, Any]:
"""
@ -429,6 +395,79 @@ class DeviceCoordinator:
"""
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:
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:
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:
"""
重启指定设备

View File

@ -344,7 +344,8 @@ class FemtoBoltManager(BaseDevice):
if not self._start_device():
raise Exception("设备启动失败")
self.is_connected = True
# 使用set_connected方法启动连接监控线程
self.set_connected(True)
self.device_info.update({
'color_resolution': self.color_resolution,
'depth_mode': self.depth_mode,
@ -357,7 +358,8 @@ class FemtoBoltManager(BaseDevice):
except Exception as e:
self.logger.error(f"FemtoBolt初始化失败: {e}")
self.is_connected = False
# 使用set_connected方法停止连接监控线程
self.set_connected(False)
self._cleanup_device()
return False
@ -652,6 +654,9 @@ class FemtoBoltManager(BaseDevice):
ret, depth_image = capture.get_depth_image()
if ret and depth_image is not None:
# 更新心跳时间,防止连接监控线程判定为超时
self.update_heartbeat()
# 根据配置选择不同的等高线生成方法
if self.algorithm_type == 'plt':
depth_colored_final = self._generate_contour_image_plt(depth_image)

View File

@ -33,17 +33,24 @@ logger = logging.getLogger(__name__)
class RealIMUDevice:
"""真实IMU设备通过串口读取姿态数据"""
def __init__(self, port, baudrate):
self.port = port
self.baudrate = baudrate
self.ser = None
self.buffer = bytearray()
self.calibration_data = None
# 串口通信配置
self.port = port # 串口端口号如COM3、/dev/ttyUSB0等
self.baudrate = baudrate # 波特率通常为9600或115200
self.ser = None # 串口连接对象,初始为空
# 数据缓冲区和校准相关
self.buffer = bytearray() # 接收数据的缓冲区
self.calibration_data = None # 校准数据,用于修正传感器偏差
# 头部姿态偏移量,用于校准初始姿态
self.head_pose_offset = {'rotation': 0, 'tilt': 0, 'pitch': 0}
# 最后一次读取的IMU数据
self.last_data = {
'roll': 0.0,
'pitch': 0.0,
'yaw': 0.0,
'temperature': 25.0
'roll': 0.0, # 横滚角绕X轴旋转
'pitch': 0.0, # 俯仰角绕Y轴旋转
'yaw': 0.0, # 偏航角绕Z轴旋转
'temperature': 25.0 # 传感器温度
}
logger.debug(f'RealIMUDevice 初始化: port={self.port}, baudrate={self.baudrate}')
self._connect()
@ -191,64 +198,6 @@ class RealIMUDevice:
pass
class MockIMUDevice:
"""模拟IMU设备"""
def __init__(self):
self.noise_level = 0.1
self.calibration_data = None # 校准数据
self.head_pose_offset = {'rotation': 0, 'tilt': 0, 'pitch': 0} # 头部姿态零点偏移
def set_calibration(self, calibration: Dict[str, Any]):
"""设置校准数据"""
self.calibration_data = calibration
if 'head_pose_offset' in calibration:
self.head_pose_offset = calibration['head_pose_offset']
def apply_calibration(self, raw_data: Dict[str, Any]) -> Dict[str, Any]:
"""应用校准:将当前姿态减去初始偏移,得到相对姿态"""
if not raw_data or 'head_pose' not in raw_data:
return raw_data
calibrated_data = raw_data.copy()
head_pose = raw_data['head_pose'].copy()
head_pose['rotation'] = head_pose['rotation'] - self.head_pose_offset['rotation']
head_pose['tilt'] = head_pose['tilt'] - self.head_pose_offset['tilt']
head_pose['pitch'] = head_pose['pitch'] - self.head_pose_offset['pitch']
calibrated_data['head_pose'] = head_pose
return calibrated_data
def read_data(self, apply_calibration: bool = True) -> Dict[str, Any]:
"""读取IMU数据"""
# 生成头部姿态角度数据,角度范围(-90°, +90°)
# 使用正弦波模拟自然的头部运动,添加随机噪声
import time
current_time = time.time()
# 旋转角(左旋为负,右旋为正)
rotation_angle = 30 * np.sin(current_time * 0.5) + np.random.normal(0, self.noise_level * 5)
rotation_angle = np.clip(rotation_angle, -90, 90)
# 倾斜角(左倾为负,右倾为正)
tilt_angle = 20 * np.sin(current_time * 0.3 + np.pi/4) + np.random.normal(0, self.noise_level * 5)
tilt_angle = np.clip(tilt_angle, -90, 90)
# 俯仰角(俯角为负,仰角为正)
pitch_angle = 15 * np.sin(current_time * 0.7 + np.pi/2) + np.random.normal(0, self.noise_level * 5)
pitch_angle = np.clip(pitch_angle, -90, 90)
# 生成原始数据
raw_data = {
'head_pose': {
'rotation': rotation_angle, # 旋转角:左旋(-), 右旋(+)
'tilt': tilt_angle, # 倾斜角:左倾(-), 右倾(+)
'pitch': pitch_angle # 俯仰角:俯角(-), 仰角(+)
},
'timestamp': datetime.now().isoformat()
}
# 应用校准并返回
return self.apply_calibration(raw_data) if apply_calibration else raw_data
class BleIMUDevice:
"""蓝牙IMU设备基于bleak实现解析逻辑参考tests/testblueimu.py"""
@ -566,7 +515,8 @@ class IMUManager(BaseDevice):
self.logger.info(f"使用蓝牙IMU设备 - MAC: {self.mac_address}")
self.imu_device = BleIMUDevice(self.mac_address)
self.imu_device.start()
self.is_connected = True
# 使用set_connected方法来正确启动连接监控线程
self.set_connected(True)
elif self.device_type == 'real' or (self.device_type != 'mock' and not self.use_mock):
self.logger.info(f"使用真实IMU设备 - 端口: {self.port}, 波特率: {self.baudrate}")
self.imu_device = RealIMUDevice(self.port, self.baudrate)
@ -577,11 +527,13 @@ class IMUManager(BaseDevice):
self.is_connected = False
self.imu_device = None
return False
self.is_connected = True
# 使用set_connected方法来正确启动连接监控线程
self.set_connected(True)
else:
self.logger.info("使用模拟IMU设备")
self.imu_device = MockIMUDevice()
self.is_connected = True
# 使用set_connected方法来正确启动连接监控线程
self.set_connected(True)
self._device_info.update({
'port': self.port,
@ -748,6 +700,9 @@ class IMUManager(BaseDevice):
# self.data_buffer.append(data)
# self.last_valid_data = data
# 更新心跳时间,防止连接监控线程判定为超时
self.update_heartbeat()
# 发送数据到前端
if self._socketio:
self._socketio.emit('imu_data', data, namespace='/devices')
@ -800,41 +755,7 @@ class IMUManager(BaseDevice):
"""
return self.last_valid_data.copy() if self.last_valid_data else None
def collect_head_pose_data(self, duration: int = 10) -> List[Dict[str, Any]]:
"""
收集头部姿态数据
Args:
duration: 收集时长
Returns:
List[Dict[str, Any]]: 收集到的数据列表
"""
collected_data = []
if not self.is_connected or not self.imu_device:
self.logger.error("IMU设备未连接")
return collected_data
self.logger.info(f"开始收集头部姿态数据,时长: {duration}")
start_time = time.time()
while time.time() - start_time < duration:
try:
data = self.imu_device.read_data(apply_calibration=True)
if data:
# 添加时间戳
data['timestamp'] = time.time()
collected_data.append(data)
time.sleep(0.02) # 50Hz采样率
except Exception as e:
self.logger.error(f"数据收集异常: {e}")
break
self.logger.info(f"头部姿态数据收集完成,共收集 {len(collected_data)} 个样本")
return collected_data
def disconnect(self):
"""
@ -900,29 +821,37 @@ class IMUManager(BaseDevice):
if not self.imu_device:
return False
# 对于真实设备,检查串口连接状态
if hasattr(self.imu_device, 'ser') and self.imu_device.ser:
# 检查串口是否仍然打开
if not self.imu_device.ser.is_open:
# 检查设备类型并分别处理
if isinstance(self.imu_device, RealIMUDevice):
# 对于真实串口设备,检查串口连接状态
if hasattr(self.imu_device, 'ser') and self.imu_device.ser:
# 检查串口是否仍然打开
if not self.imu_device.ser.is_open:
return False
# 尝试读取数据来验证连接
try:
# 保存当前超时设置
original_timeout = self.imu_device.ser.timeout
self.imu_device.ser.timeout = 0.1 # 设置短超时
# 尝试读取少量数据
test_data = self.imu_device.ser.read(1)
# 恢复原始超时设置
self.imu_device.ser.timeout = original_timeout
return True # 如果没有异常,认为连接正常
except Exception:
return False
else:
return False
# 尝试读取数据来验证连接
try:
# 保存当前超时设置
original_timeout = self.imu_device.ser.timeout
self.imu_device.ser.timeout = 0.1 # 设置短超时
elif isinstance(self.imu_device, BleIMUDevice):
# 对于蓝牙设备,检查连接状态
return self.imu_device.connected
# 尝试读取少量数据
test_data = self.imu_device.ser.read(1)
# 恢复原始超时设置
self.imu_device.ser.timeout = original_timeout
return True # 如果没有异常,认为连接正常
except Exception:
return False
# 对于模拟设备总是返回True
# 对于模拟设备或其他类型总是返回True
return True
except Exception as e:

View File

@ -167,6 +167,7 @@ class RealPressureDevice:
self.frame_size = self.rows * self.cols
self.buf_type = ctypes.c_uint16 * self.frame_size
self.buf = self.buf_type()
# 设置连接状态
self.is_connected = True
logger.info(f"SMiTSense压力传感器初始化成功: {self.rows}行 x {self.cols}")
@ -179,7 +180,9 @@ class RealPressureDevice:
"""读取压力数据并转换为与MockPressureDevice兼容的格式"""
try:
if not self.is_connected or not self.dll:
logger.error("设备未连接")
return self._get_empty_data()
# 检查device_handle是否有效
if not self.device_handle:
return self._get_empty_data()
# 读取原始压力数据
@ -194,10 +197,10 @@ class RealPressureDevice:
self.dll.fpms_usb_close_wrap(self.device_handle.value)
except Exception:
pass
self.device_handle = None
finally:
self.device_handle = None
except Exception:
pass
self.is_connected = False
return self._get_empty_data()
# 转换为numpy数组
@ -430,6 +433,7 @@ class RealPressureDevice:
try:
if self.is_connected and self.dll and self.device_handle:
self.dll.fpms_usb_close_wrap(self.device_handle.value)
# 设置连接状态为断开
self.is_connected = False
logger.info('SMiTSense压力传感器连接已关闭')
except Exception as e:
@ -440,267 +444,6 @@ class RealPressureDevice:
self.close()
class MockPressureDevice:
"""模拟压力传感器设备模拟真实SMiTSense设备的行为"""
def __init__(self):
self.base_pressure = 500 # 基础压力值
self.noise_level = 10
self.rows = 4 # 模拟传感器矩阵行数
self.cols = 4 # 模拟传感器矩阵列数
self.time_offset = np.random.random() * 10 # 随机时间偏移,让每个实例的波形不同
def read_data(self) -> Dict[str, Any]:
"""读取压力数据,模拟基于矩阵数据的真实设备行为"""
try:
# 生成模拟的传感器矩阵数据
raw_data = self._generate_simulated_matrix_data()
# 使用与真实设备相同的计算逻辑
foot_zones = self._calculate_foot_pressure_zones(raw_data)
# 生成压力图像
pressure_image_base64 = self._generate_pressure_image(
foot_zones['left_front'],
foot_zones['left_rear'],
foot_zones['right_front'],
foot_zones['right_rear'],
raw_data
)
return {
'foot_pressure': {
'left_front': round(foot_zones['left_front'], 2),
'left_rear': round(foot_zones['left_rear'], 2),
'right_front': round(foot_zones['right_front'], 2),
'right_rear': round(foot_zones['right_rear'], 2),
'left_total': round(foot_zones['left_total'], 2),
'right_total': round(foot_zones['right_total'], 2)
},
'pressure_image': pressure_image_base64,
'timestamp': datetime.now().isoformat()
}
except Exception as e:
logger.error(f"模拟压力设备读取数据异常: {e}")
return self._get_empty_data()
def _generate_simulated_matrix_data(self):
"""生成模拟的传感器矩阵数据,模拟真实的足部压力分布"""
import time
current_time = time.time() + self.time_offset
# 创建4x4的传感器矩阵
matrix_data = np.zeros((self.rows, self.cols))
# 模拟动态的压力分布,使用正弦波叠加噪声
for i in range(self.rows):
for j in range(self.cols):
# 基础压力值,根据传感器位置不同
base_value = self.base_pressure * (0.3 + 0.7 * np.random.random())
# 添加时间变化(模拟人体重心变化)
time_variation = np.sin(current_time * 0.5 + i * 0.5 + j * 0.3) * 0.3
# 添加噪声
noise = np.random.normal(0, self.noise_level)
# 确保压力值非负
matrix_data[i, j] = max(0, base_value * (1 + time_variation) + noise)
return matrix_data
def _calculate_foot_pressure_zones(self, raw_data):
"""计算足部区域压力,返回百分比:
- 左足右足相对于双足总压的百分比
- 左前左后相对于左足总压的百分比
- 右前右后相对于右足总压的百分比
基于原始矩阵按行列各等分为四象限上半部为前下半部为后左半部为左右半部为右
"""
try:
# 防护:空数据
if raw_data is None:
raise ValueError("raw_data is None")
# 转为浮点以避免 uint16 溢出
rd = np.asarray(raw_data, dtype=np.float64)
rows, cols = rd.shape if rd.ndim == 2 else (0, 0)
if rows == 0 or cols == 0:
raise ValueError("raw_data has invalid shape")
# 行列对半分(上=前,下=后;左=左,右=右)
mid_r = rows // 2
mid_c = cols // 2
# 四象限求和
left_front = float(np.sum(rd[:mid_r, :mid_c], dtype=np.float64))
left_rear = float(np.sum(rd[mid_r:, :mid_c], dtype=np.float64))
right_front = float(np.sum(rd[:mid_r, mid_c:], dtype=np.float64))
right_rear = float(np.sum(rd[mid_r:, mid_c:], dtype=np.float64))
# 绝对总压
left_total_abs = left_front + left_rear
right_total_abs = right_front + right_rear
total_abs = left_total_abs + right_total_abs
# 左右足占比(相对于双足总压)
left_total_pct = float((left_total_abs / total_abs * 100) if total_abs > 0 else 0)
right_total_pct = float((right_total_abs / total_abs * 100) if total_abs > 0 else 0)
# 前后占比(相对于各自单足总压)
left_front_pct = float((left_front / left_total_abs * 100) if left_total_abs > 0 else 0)
left_rear_pct = float((left_rear / left_total_abs * 100) if left_total_abs > 0 else 0)
right_front_pct = float((right_front / right_total_abs * 100) if right_total_abs > 0 else 0)
right_rear_pct = float((right_rear / right_total_abs * 100) if right_total_abs > 0 else 0)
return {
'left_front': left_front_pct,
'left_rear': left_rear_pct,
'right_front': right_front_pct,
'right_rear': right_rear_pct,
'left_total': left_total_pct,
'right_total': right_total_pct,
'total_pressure': float(total_abs)
}
except Exception as e:
logger.error(f"计算足部区域压力异常: {e}")
return {
'left_front': 0, 'left_rear': 0, 'right_front': 0, 'right_rear': 0,
'left_total': 0, 'right_total': 0, 'total_pressure': 0
}
def _generate_pressure_image(self, left_front, left_rear, right_front, right_rear, raw_data=None) -> str:
"""生成足部压力图片的base64数据"""
try:
if MATPLOTLIB_AVAILABLE and raw_data is not None:
# 使用原始数据生成更详细的热力图
return self._generate_heatmap_image(raw_data)
else:
# 降级到简单的区域显示图
return self._generate_simple_pressure_image(left_front, left_rear, right_front, right_rear)
except Exception as e:
logger.warning(f"生成模拟压力图片失败: {e}")
return "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
def _generate_heatmap_image(self, raw_data) -> str:
"""生成基于原始数据的热力图"""
try:
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from io import BytesIO
# 参考 tests/testsmit.py 的渲染方式:使用 jet 色图、nearest 插值、固定范围并关闭坐标轴
fig, ax = plt.subplots()
im = ax.imshow(raw_data, cmap='jet', interpolation='nearest', vmin=0, vmax=1000)
ax.axis('off')
# 紧凑布局并导出为 base64
from io import BytesIO
buffer = BytesIO()
plt.savefig(buffer, format='png', bbox_inches='tight', dpi=100, pad_inches=0)
buffer.seek(0)
image_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
plt.close(fig)
return f"data:image/png;base64,{image_base64}"
except Exception as e:
logger.warning(f"生成热力图失败: {e}")
return self._generate_simple_pressure_image(0, 0, 0, 0)
def _generate_simple_pressure_image(self, left_front, left_rear, right_front, right_rear) -> str:
"""生成简单的足部压力区域图"""
try:
import matplotlib
matplotlib.use('Agg') # 设置非交互式后端避免Tkinter错误
import matplotlib.pyplot as plt
import matplotlib.patches as patches
from io import BytesIO
# 临时禁用PIL的调试日志
pil_logger = logging.getLogger('PIL')
original_level = pil_logger.level
pil_logger.setLevel(logging.WARNING)
# 创建图形
fig, ax = plt.subplots(1, 1, figsize=(6, 8))
ax.set_xlim(0, 10)
ax.set_ylim(0, 12)
ax.set_aspect('equal')
ax.axis('off')
# 定义颜色映射(根据压力值)
max_pressure = max(left_front, left_rear, right_front, right_rear)
if max_pressure > 0:
left_front_color = plt.cm.Reds(left_front / max_pressure)
left_rear_color = plt.cm.Reds(left_rear / max_pressure)
right_front_color = plt.cm.Reds(right_front / max_pressure)
right_rear_color = plt.cm.Reds(right_rear / max_pressure)
else:
left_front_color = left_rear_color = right_front_color = right_rear_color = 'lightgray'
# 绘制左脚
left_front_rect = patches.Rectangle((1, 6), 2, 4, linewidth=1, edgecolor='black', facecolor=left_front_color)
left_rear_rect = patches.Rectangle((1, 2), 2, 4, linewidth=1, edgecolor='black', facecolor=left_rear_color)
# 绘制右脚
right_front_rect = patches.Rectangle((7, 6), 2, 4, linewidth=1, edgecolor='black', facecolor=right_front_color)
right_rear_rect = patches.Rectangle((7, 2), 2, 4, linewidth=1, edgecolor='black', facecolor=right_rear_color)
# 添加到图形
ax.add_patch(left_front_rect)
ax.add_patch(left_rear_rect)
ax.add_patch(right_front_rect)
ax.add_patch(right_rear_rect)
# 添加标签
ax.text(2, 8, f'{left_front:.1f}', ha='center', va='center', fontsize=10, weight='bold')
ax.text(2, 4, f'{left_rear:.1f}', ha='center', va='center', fontsize=10, weight='bold')
ax.text(8, 8, f'{right_front:.1f}', ha='center', va='center', fontsize=10, weight='bold')
ax.text(8, 4, f'{right_rear:.1f}', ha='center', va='center', fontsize=10, weight='bold')
ax.text(2, 0.5, '左足', ha='center', va='center', fontsize=12, weight='bold')
ax.text(8, 0.5, '右足', ha='center', va='center', fontsize=12, weight='bold')
# 保存为base64
buffer = BytesIO()
plt.savefig(buffer, format='png', bbox_inches='tight', dpi=100, facecolor='white')
buffer.seek(0)
image_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
plt.close(fig)
# 恢复PIL的日志级别
pil_logger.setLevel(original_level)
return f"data:image/png;base64,{image_base64}"
except Exception as e:
# 确保在异常情况下也恢复PIL的日志级别
try:
pil_logger.setLevel(original_level)
except:
pass
logger.warning(f"生成压力图片失败: {e}")
# 返回一个简单的占位符base64图片
return "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
def _get_empty_data(self):
"""返回空的压力数据"""
return {
'foot_pressure': {
'left_front': 0.0,
'left_rear': 0.0,
'right_front': 0.0,
'right_rear': 0.0,
'left_total': 0.0,
'right_total': 0.0
},
'pressure_image': "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==",
'timestamp': datetime.now().isoformat()
}
class PressureManager(BaseDevice):
"""压力板管理器"""
@ -767,7 +510,8 @@ class PressureManager(BaseDevice):
else:
self.device = MockPressureDevice()
self.is_connected = True
# 使用set_connected方法启动连接监控线程
self.set_connected(True)
self._device_info.update({
'device_type': self.device_type,
'matrix_size': '4x4' if hasattr(self.device, 'rows') else 'unknown'
@ -843,54 +587,12 @@ class PressureManager(BaseDevice):
"""
self.logger.info("压力数据流线程启动")
reconnect_attempts = 0
consecutive_read_failures = 0
try:
while self.is_streaming:
try:
# 若设备未连接或不存在,进入重连流程
if not self.device or not self.is_connected or (hasattr(self.device, 'is_connected') and not self.device.is_connected):
# 广播断开状态(仅状态变化时)
if self._last_connected_state is not False:
try:
if self._socketio:
self._socketio.emit('pressure_status', {
'status': 'disconnected',
'timestamp': time.time()
}, namespace='/devices')
except Exception:
pass
self._last_connected_state = False
if self.max_reconnect_attempts == -1 or reconnect_attempts < self.max_reconnect_attempts:
self.logger.warning(f"压力设备连接丢失,尝试重连 ({'' if self.max_reconnect_attempts == -1 else reconnect_attempts + 1}/{self.max_reconnect_attempts if self.max_reconnect_attempts != -1 else ''})")
if not self.is_streaming:
break
if self._reconnect():
reconnect_attempts = 0
consecutive_read_failures = 0
# 广播恢复
try:
if self._socketio:
self._socketio.emit('pressure_status', {
'status': 'connected',
'timestamp': time.time()
}, namespace='/devices')
except Exception:
pass
self._last_connected_state = True
continue
else:
reconnect_attempts += 1
time.sleep(self.reconnect_delay)
continue
else:
self.logger.error("压力设备重连失败次数过多,进入降频重试模式")
time.sleep(max(self.reconnect_delay, 5.0))
reconnect_attempts = 0
continue
# 读数成功,重置失败计数
self.is_connected = True
# 从设备读取数据
pressure_data = None
if self.device:
@ -901,26 +603,6 @@ class PressureManager(BaseDevice):
time.sleep(self.reconnect_delay)
continue
if not pressure_data or 'foot_pressure' not in pressure_data:
consecutive_read_failures += 1
if consecutive_read_failures >= self.read_fail_threshold:
self.logger.warning(f"连续读取压力数据失败 {consecutive_read_failures} 次,执行设备软复位并进入重连")
try:
if self.device and hasattr(self.device, 'close'):
self.device.close()
except Exception:
pass
self.is_connected = False
consecutive_read_failures = 0
time.sleep(self.reconnect_delay)
continue
time.sleep(0.05)
continue
# 读数成功,重置失败计数
consecutive_read_failures = 0
self.is_connected = True
foot_pressure = pressure_data['foot_pressure']
# 获取各区域压力值
left_front = foot_pressure['left_front']
@ -965,6 +647,9 @@ class PressureManager(BaseDevice):
'timestamp': pressure_data['timestamp']
}
# 更新心跳时间,防止连接监控线程判定为超时
self.update_heartbeat()
# 更新统计信息
self.packet_count += 1
self.last_data_time = time.time()
@ -982,7 +667,7 @@ class PressureManager(BaseDevice):
except Exception as e:
self.error_count += 1
self.logger.error(f"压力数据流处理异常: {e}")
# self.logger.error(f"压力数据流处理异常: {e}")
time.sleep(0.1)
except Exception as e:
@ -990,41 +675,7 @@ class PressureManager(BaseDevice):
finally:
self.logger.info("压力数据流线程结束")
def _reconnect(self) -> bool:
"""重新连接压力设备"""
try:
# 先清理旧设备
try:
if self.device and hasattr(self.device, 'close'):
self.device.close()
except Exception:
pass
self.device = None
self.is_connected = False
time.sleep(1.0) # 等待设备释放
# 重新创建设备
if self.device_type == 'real':
self.device = RealPressureDevice()
else:
self.device = MockPressureDevice()
self.is_connected = True
# 广播一次连接状态
try:
if self._socketio:
self._socketio.emit('pressure_status', {
'status': 'connected',
'timestamp': time.time()
}, namespace='/devices')
except Exception:
pass
return True
except Exception as e:
self.logger.error(f"压力设备重连失败: {e}")
self.is_connected = False
return False
def get_status(self) -> Dict[str, Any]:
"""
@ -1085,7 +736,8 @@ class PressureManager(BaseDevice):
self.device.close()
self.device = None
self.is_connected = False
# 使用set_connected方法停止连接监控线程
self.set_connected(False)
self.logger.info("压力板设备连接已断开")
return True
@ -1132,23 +784,29 @@ class PressureManager(BaseDevice):
"""
try:
if not self.device:
return False
# 如果设备实例不存在,尝试重新创建
return self._attempt_device_reconnection()
# 对于真实设备检查DLL和设备句柄状态
if hasattr(self.device, 'dll') and hasattr(self.device, 'device_handle'):
if not self.device.dll or not self.device.device_handle:
return False
# DLL或句柄无效尝试重连
return self._attempt_device_reconnection()
# 检查设备连接状态
if not self.device.is_connected:
return False
# 尝试读取一次数据来验证连接
# 直接检查设备句柄的有效性不依赖is_connected状态避免循环依赖
try:
test_data = self.device.read_data()
return test_data is not None and 'foot_pressure' in test_data
# 检查设备句柄是否有效
if not self.device.device_handle or not hasattr(self.device.device_handle, 'value'):
return self._attempt_device_reconnection()
# 检查句柄值是否为0无效句柄
if self.device.device_handle.value == 0:
return self._attempt_device_reconnection()
# 尝试简单的设备状态检查,而不是完整的数据读取
return True
except Exception:
return False
return self._attempt_device_reconnection()
# 对于模拟设备总是返回True
return True
@ -1157,6 +815,49 @@ class PressureManager(BaseDevice):
self.logger.debug(f"检查压力板硬件连接时出错: {e}")
return False
def _attempt_device_reconnection(self) -> bool:
"""
尝试重新连接压力板设备
Returns:
bool: 重连是否成功
"""
try:
self.logger.info("检测到压力板设备断开,尝试重新连接...")
# 清理旧的设备实例
if self.device and hasattr(self.device, 'close'):
try:
self.device.close()
except Exception as e:
self.logger.debug(f"清理旧设备实例时出错: {e}")
self.device = None
# 根据设备类型重新创建设备实例
if self.device_type == 'real':
self.device = RealPressureDevice()
else:
self.device = MockPressureDevice()
# 检查新设备是否连接成功
if hasattr(self.device, 'is_connected') and self.device.is_connected:
self._notify_status_change(True)
# 重连成功后,确保数据流正在运行
if not self.is_streaming:
self.logger.info("重连成功,启动压力数据流")
self.start_streaming()
return True
else:
self.logger.warning("压力板设备重连失败")
return False
except Exception as e:
self.logger.error(f"压力板设备重连过程中出错: {e}")
self.device = None
return False
def cleanup(self) -> None:
"""清理资源"""
try:

View File

@ -15,14 +15,17 @@ backup_interval = 24
max_backups = 7
[CAMERA]
device_index = 3
enabled = True
device_index = 0
width = 1280
height = 720
fps = 30
buffer_size = 1
fourcc = MJPG
backend = directshow
[FEMTOBOLT]
enabled = True
algorithm_type = opencv
color_resolution = 1080P
depth_mode = NFOV_2X2BINNED
@ -33,10 +36,13 @@ fps = 15
synchronized_images_only = False
[DEVICES]
imu_enabled = True
imu_device_type = ble
imu_port = COM9
imu_mac_address = ef:3c:1a:0a:fe:02
imu_baudrate = 9600
pressure_enabled = True
pressure_device_type = real
pressure_use_mock = False
pressure_port = COM5

View File

@ -109,7 +109,8 @@ class ConfigManager:
'device_index': '0',
'width': '1280',
'height': '720',
'fps': '30'
'fps': '30',
'backend': 'directshow'
}
# 默认FemtoBolt配置
@ -168,12 +169,14 @@ class ConfigManager:
Dict[str, Any]: 相机配置
"""
return {
'enabled': self.config.getboolean('CAMERA', 'enabled', fallback=True),
'device_index': self.config.getint('CAMERA', 'device_index', fallback=0),
'width': self.config.getint('CAMERA', 'width', fallback=1280),
'height': self.config.getint('CAMERA', 'height', fallback=720),
'fps': self.config.getint('CAMERA', 'fps', fallback=30),
'buffer_size': self.config.getint('CAMERA', 'buffer_size', fallback=1),
'fourcc': self.config.get('CAMERA', 'fourcc', fallback='MJPG')
'fourcc': self.config.get('CAMERA', 'fourcc', fallback='MJPG'),
'backend': self.config.get('CAMERA', 'backend', fallback='directshow')
}
def _get_femtobolt_config(self) -> Dict[str, Any]:
@ -184,6 +187,7 @@ class ConfigManager:
Dict[str, Any]: FemtoBolt配置
"""
return {
'enabled': self.config.getboolean('FEMTOBOLT', 'enabled', fallback=True),
'algorithm_type': self.config.get('FEMTOBOLT', 'algorithm_type', fallback='opencv'),
'color_resolution': self.config.get('FEMTOBOLT', 'color_resolution', fallback='1080P'),
'depth_mode': self.config.get('FEMTOBOLT', 'depth_mode', fallback='NFOV_UNBINNED'),
@ -201,6 +205,7 @@ class ConfigManager:
Dict[str, Any]: IMU配置
"""
return {
'enabled': self.config.getboolean('DEVICES', 'imu_enabled', fallback=True),
'device_type': self.config.get('DEVICES', 'imu_device_type', fallback='mock'),
'port': self.config.get('DEVICES', 'imu_port', fallback='COM7'),
'baudrate': self.config.getint('DEVICES', 'imu_baudrate', fallback=9600),
@ -217,6 +222,7 @@ class ConfigManager:
Dict[str, Any]: 压力传感器配置
"""
return {
'enabled': self.config.getboolean('DEVICES', 'pressure_enabled', fallback=True),
'device_type': self.config.get('DEVICES', 'pressure_device_type', fallback='mock'),
'port': self.config.get('DEVICES', 'pressure_port', fallback='COM8'),
'baudrate': self.config.getint('DEVICES', 'pressure_baudrate', fallback=115200),
@ -458,6 +464,8 @@ class ConfigManager:
self.set_config_value('CAMERA', 'height', str(config_data['height']))
if 'fps' in config_data:
self.set_config_value('CAMERA', 'fps', str(config_data['fps']))
if 'backend' in config_data:
self.set_config_value('CAMERA', 'backend', str(config_data['backend']))
# 保存配置
self.save_config()
@ -619,6 +627,8 @@ class ConfigManager:
self.set_config_value('CAMERA', 'fourcc', config_data['fourcc'])
if 'tx_max_width' in config_data:
self.set_config_value('CAMERA', 'tx_max_width', str(config_data['tx_max_width']))
if 'backend' in config_data:
self.set_config_value('CAMERA', 'backend', str(config_data['backend']))
results['camera'] = {
'success': True,

View File

@ -184,34 +184,37 @@ class AppServer:
self.config_manager = ConfigManager()
self.logger.info('配置管理器初始化完成')
# 初始化设备管理器
self.logger.info('正在初始化设备管理器...')
self.device_managers = {
'camera': CameraManager(self.socketio, self.config_manager),
'femtobolt': FemtoBoltManager(self.socketio, self.config_manager),
'imu': IMUManager(self.socketio, self.config_manager),
'pressure': PressureManager(self.socketio, self.config_manager)
}
# 为每个设备添加状态变化回调
for device_name, manager in self.device_managers.items():
if manager and hasattr(manager, 'add_status_change_callback'):
manager.add_status_change_callback(self._on_device_status_change)
self.logger.info('设备管理器初始化完成')
# 初始化设备协调器
# 初始化设备协调器(统一管理所有设备)
self.logger.info('正在初始化设备协调器...')
self.device_coordinator = DeviceCoordinator(self.socketio)
self.logger.info('设备协调器初始化完成')
# 调用初始化方法来初始化设备
if self.device_coordinator.initialize():
self.logger.info('设备协调器初始化完成')
# 获取设备管理器实例
self.device_managers = self.device_coordinator.get_device_managers()
# 为每个设备添加状态变化回调
for device_name, manager in self.device_managers.items():
if manager and hasattr(manager, 'add_status_change_callback'):
manager.add_status_change_callback(self._on_device_status_change)
self.logger.info(f'已获取设备管理器: {list(self.device_managers.keys())}')
else:
self.logger.warning('设备协调器初始化失败,但系统将继续运行')
self.device_managers = {} # 初始化为空字典以避免后续错误
# 初始化录制管理器
self.logger.info('正在初始化录制管理器...')
self.recording_manager = RecordingManager(
camera_manager=self.device_managers['camera'],
db_manager=self.db_manager
)
self.logger.info('录制管理器初始化完成')
camera_manager = self.device_managers.get('camera')
if camera_manager:
self.recording_manager = RecordingManager(
camera_manager=camera_manager,
db_manager=self.db_manager
)
self.logger.info('录制管理器初始化完成')
else:
self.recording_manager = None
self.logger.warning('相机设备未初始化,录制管理器将不可用')
# 启动Flask应用
host = self.host
@ -985,6 +988,15 @@ class AppServer:
# 调用create_detection_session方法settings传空字典
session_id = self.db_manager.create_detection_session(patient_id, settings={}, creator_id=creator_id)
self.logger.info('检测开始,设备连接监控已启动')
# 启动设备连接监控
if self.device_coordinator:
try:
self.device_coordinator.start_all_connection_monitor()
self.logger.info('检测开始,设备连接监控已启动')
except Exception as monitor_error:
self.logger.error(f'启动设备连接监控失败: {monitor_error}')
return jsonify({'success': True, 'session_id': session_id})
except Exception as e:
self.logger.error(f'开始检测失败: {e}')
@ -1017,6 +1029,14 @@ class AppServer:
except Exception as duration_error:
self.logger.error(f'更新会话持续时间失败: {duration_error}')
# 停止设备连接监控
if self.device_coordinator:
try:
self.device_coordinator.stop_all_connection_monitor()
self.logger.info('检测停止,设备连接监控已停止')
except Exception as monitor_error:
self.logger.error(f'停止设备连接监控失败: {monitor_error}')
success = self.db_manager.update_session_status(session_id, 'completed')
if success:
self.logger.info(f'检测会话已停止 - 会话ID: {session_id}')
@ -1508,6 +1528,17 @@ class AppServer:
def initialize_device(device_name, manager):
"""设备初始化工作函数"""
try:
# 检查设备是否已连接,避免重复初始化
if hasattr(manager, 'is_connected') and manager.is_connected:
print(f"[DEBUG] {device_name} 已连接,跳过初始化")
# 如果已连接但未启动流,则启动流
if hasattr(manager, 'is_streaming') and not manager.is_streaming:
print(f"[DEBUG] {device_name} 已连接但未启动流,开始启动流")
manager.start_streaming()
device_results[device_name] = True
self.logger.info(f'{device_name}设备已连接,启动成功')
return
print(f"[DEBUG] 尝试初始化设备: {device_name}")
if manager.initialize():
print(f"[DEBUG] {device_name} 初始化成功,开始启动流")

View File

@ -25,8 +25,9 @@ Pillow
# Data visualization and report generation
reportlab
# Serial communication
# Serial communication and Bluetooth
pyserial
bleak # Bluetooth Low Energy communication for IMU devices
# Audio/video processing
ffmpeg-python

View File

@ -0,0 +1,133 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
简化的相机性能测试
"""
import sys
import os
import time
import logging
# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from devices.utils.config_manager import ConfigManager
from devices.camera_manager import CameraManager
# 设置日志级别
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def test_camera_init_time():
"""
测试相机初始化时间
"""
print("相机初始化性能测试")
print("=" * 50)
try:
# 创建管理器
config_manager = ConfigManager()
camera_manager = CameraManager(None, config_manager)
print("\n开始相机初始化测试...")
# 记录总时间
total_start = time.time()
# 执行初始化
success = camera_manager.initialize()
total_time = (time.time() - total_start) * 1000
print(f"\n初始化结果: {'成功' if success else '失败'}")
print(f"总耗时: {total_time:.1f}ms ({total_time/1000:.1f}秒)")
# 性能评估
if total_time < 1000:
print("性能评级: 优秀 ⭐⭐⭐ (< 1秒)")
elif total_time < 3000:
print("性能评级: 良好 ⭐⭐ (< 3秒)")
elif total_time < 5000:
print("性能评级: 一般 ⭐ (< 5秒)")
else:
print("性能评级: 需要优化 ❌ (> 5秒)")
if success:
# 测试校准
print("\n测试校准性能...")
calibrate_start = time.time()
calibrate_success = camera_manager.calibrate()
calibrate_time = (time.time() - calibrate_start) * 1000
print(f"校准结果: {'成功' if calibrate_success else '失败'}")
print(f"校准耗时: {calibrate_time:.1f}ms")
# 测试首帧获取
if camera_manager.cap:
print("\n测试首帧获取...")
frame_start = time.time()
ret, frame = camera_manager.cap.read()
frame_time = (time.time() - frame_start) * 1000
if ret and frame is not None:
print(f"首帧获取成功 - 耗时: {frame_time:.1f}ms, 帧大小: {frame.shape}")
del frame
else:
print(f"首帧获取失败 - 耗时: {frame_time:.1f}ms")
# 获取设备信息
print("\n设备信息:")
device_info = camera_manager.get_device_info()
for key, value in device_info.items():
if key in ['width', 'height', 'fps', 'fourcc']:
print(f" {key}: {value}")
# 清理
camera_manager.cleanup()
except Exception as e:
print(f"\n❌ 测试失败: {e}")
import traceback
traceback.print_exc()
def analyze_performance_bottlenecks():
"""
分析性能瓶颈
"""
print("\n" + "=" * 50)
print("性能瓶颈分析")
print("=" * 50)
print("\n根据测试结果,主要性能瓶颈可能包括:")
print("1. 相机打开 (CAP_PROP设置) - 通常耗时3-4秒")
print("2. 分辨率设置 - 可能耗时5-6秒")
print("3. FPS设置 - 可能耗时2-3秒")
print("4. 首帧读取 - 通常耗时300-400ms")
print("\n优化建议:")
print("• 使用更快的相机后端 (如DirectShow)")
print("• 减少不必要的属性设置")
print("• 使用较低的分辨率进行初始化")
print("• 启用OpenCV优化")
print("• 设置合适的缓冲区大小")
def main():
print("相机启动性能优化测试")
print("目标: 将启动时间从10+秒优化到3秒以内")
# 执行测试
test_camera_init_time()
# 分析结果
analyze_performance_bottlenecks()
print("\n" + "=" * 50)
print("测试完成")
print("=" * 50)
if __name__ == '__main__':
main()

View File

@ -0,0 +1,75 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试避免重复初始化功能
"""
import requests
import time
import json
def test_avoid_duplicate_initialization():
"""
测试避免重复初始化功能
"""
base_url = "http://localhost:5000"
print("=== 测试避免重复初始化功能 ===")
# 1. 获取初始设备状态
print("\n1. 获取初始设备状态")
try:
response = requests.get(f"{base_url}/api/devices/status")
if response.status_code == 200:
data = response.json()
print(f"设备状态: {json.dumps(data, indent=2, ensure_ascii=False)}")
else:
print(f"获取设备状态失败: {response.status_code}")
except Exception as e:
print(f"请求失败: {e}")
# 2. 第一次启动设备数据推送
print("\n2. 第一次启动设备数据推送")
try:
response = requests.post(f"{base_url}/api/devices/start_push")
if response.status_code == 200:
data = response.json()
print(f"第一次启动结果: {json.dumps(data, indent=2, ensure_ascii=False)}")
else:
print(f"第一次启动失败: {response.status_code}")
except Exception as e:
print(f"请求失败: {e}")
# 等待一段时间
print("\n等待5秒...")
time.sleep(5)
# 3. 第二次启动设备数据推送(应该避免重复初始化)
print("\n3. 第二次启动设备数据推送(测试避免重复初始化)")
try:
response = requests.post(f"{base_url}/api/devices/start_push")
if response.status_code == 200:
data = response.json()
print(f"第二次启动结果: {json.dumps(data, indent=2, ensure_ascii=False)}")
else:
print(f"第二次启动失败: {response.status_code}")
except Exception as e:
print(f"请求失败: {e}")
# 4. 再次获取设备状态
print("\n4. 获取最终设备状态")
try:
response = requests.get(f"{base_url}/api/devices/status")
if response.status_code == 200:
data = response.json()
print(f"最终设备状态: {json.dumps(data, indent=2, ensure_ascii=False)}")
else:
print(f"获取设备状态失败: {response.status_code}")
except Exception as e:
print(f"请求失败: {e}")
print("\n=== 测试完成 ===")
print("请查看终端日志,确认第二次启动时是否显示'已连接,跳过初始化'的消息")
if __name__ == "__main__":
test_avoid_duplicate_initialization()

View File

@ -0,0 +1,235 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
OpenCV后端优化验证脚本
测试DirectShow后端相对于MSMF的性能提升
"""
import sys
import os
import time
import logging
# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from devices.camera_manager import CameraManager
from devices.utils.config_manager import ConfigManager
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_backend_performance(backend_name, test_name):
"""
测试指定后端的性能
Args:
backend_name: 后端名称 (directshow, msmf)
test_name: 测试名称
Returns:
dict: 性能数据
"""
print(f"\n{'='*60}")
print(f"📷 测试 {test_name} 后端性能")
print(f"{'='*60}")
# 创建配置管理器并设置后端
config_manager = ConfigManager()
# 获取原始配置
original_config = config_manager.get_device_config('camera')
# 设置测试后端
test_config = {
'backend': backend_name
}
config_manager.set_camera_config(test_config)
try:
# 创建相机管理器
camera = CameraManager(None, config_manager)
# 测试初始化性能
start_time = time.time()
success = camera.initialize()
total_time = (time.time() - start_time) * 1000
if success:
print(f"✅ 相机初始化成功: {total_time:.1f}ms")
# 获取实际后端信息
if camera.cap:
backend_info = camera.cap.getBackendName() if hasattr(camera.cap, 'getBackendName') else 'Unknown'
actual_width = int(camera.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_height = int(camera.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
actual_fps = camera.cap.get(cv2.CAP_PROP_FPS)
print(f"🎯 实际后端: {backend_info}")
print(f"📐 实际分辨率: {actual_width}x{actual_height}@{actual_fps:.1f}fps")
# 测试首帧获取
frame_start = time.time()
ret, frame = camera.cap.read() if camera.cap else (False, None)
frame_time = (time.time() - frame_start) * 1000
if ret and frame is not None:
print(f"🖼️ 首帧获取: {frame_time:.1f}ms, 帧大小: {frame.shape}")
else:
print(f"❌ 首帧获取失败")
frame_time = -1
# 测试连续帧性能
print(f"🎬 测试连续帧获取性能...")
frame_times = []
for i in range(10):
frame_start = time.time()
ret, frame = camera.cap.read() if camera.cap else (False, None)
if ret:
frame_times.append((time.time() - frame_start) * 1000)
time.sleep(0.01) # 小延迟
if frame_times:
avg_frame_time = sum(frame_times) / len(frame_times)
min_frame_time = min(frame_times)
max_frame_time = max(frame_times)
print(f"📈 连续帧性能: 平均 {avg_frame_time:.1f}ms, 最快 {min_frame_time:.1f}ms, 最慢 {max_frame_time:.1f}ms")
else:
avg_frame_time = -1
print(f"❌ 连续帧测试失败")
# 清理资源
camera.cleanup()
print(f"🧹 相机资源已释放")
return {
'backend': backend_name,
'success': True,
'init_time': total_time,
'first_frame_time': frame_time,
'avg_frame_time': avg_frame_time,
'backend_info': backend_info if camera.cap else 'Unknown',
'resolution': f"{actual_width}x{actual_height}@{actual_fps:.1f}fps" if camera.cap else "未知"
}
else:
print(f"❌ 初始化失败")
return {
'backend': backend_name,
'success': False,
'init_time': total_time,
'first_frame_time': -1,
'avg_frame_time': -1,
'backend_info': 'Failed',
'resolution': "失败"
}
except Exception as e:
print(f"❌ 测试异常: {e}")
return {
'backend': backend_name,
'success': False,
'init_time': -1,
'first_frame_time': -1,
'avg_frame_time': -1,
'backend_info': 'Exception',
'resolution': "异常",
'error': str(e)
}
finally:
# 恢复原始配置
try:
restore_config = {
'backend': original_config['backend']
}
config_manager.set_camera_config(restore_config)
except Exception as e:
print(f"⚠️ 恢复配置失败: {e}")
def main():
"""
主函数测试不同后端的性能
"""
print("\n" + "="*80)
print("🚀 OpenCV后端性能优化验证")
print("="*80)
# 测试用例
test_cases = [
('directshow', 'DirectShow (推荐)'),
('msmf', 'MSMF (默认)')
]
results = []
# 执行测试
for backend, name in test_cases:
result = test_backend_performance(backend, name)
results.append(result)
time.sleep(1) # 测试间隔
# 汇总结果
print(f"\n\n{'='*80}")
print(f"📊 后端性能优化验证汇总")
print(f"{'='*80}")
# 表格头
print(f"{'后端':<12} {'状态':<8} {'初始化':<12} {'首帧':<12} {'连续帧':<12} {'实际后端':<15}")
print("-" * 80)
successful_results = [r for r in results if r['success']]
for result in results:
status = "✅成功" if result['success'] else "❌失败"
init_time = f"{result['init_time']:.1f}ms" if result['init_time'] > 0 else "失败"
first_frame = f"{result['first_frame_time']:.1f}ms" if result['first_frame_time'] > 0 else "失败"
avg_frame = f"{result['avg_frame_time']:.1f}ms" if result['avg_frame_time'] > 0 else "失败"
backend_info = result['backend_info'][:14] if len(result['backend_info']) > 14 else result['backend_info']
print(f"{result['backend']:<12} {status:<8} {init_time:<12} {first_frame:<12} {avg_frame:<12} {backend_info:<15}")
# 性能分析
if len(successful_results) >= 2:
print(f"\n📈 性能分析:")
# 找到最快的后端
fastest = min(successful_results, key=lambda x: x['init_time'])
slowest = max(successful_results, key=lambda x: x['init_time'])
print(f"🏆 最快后端: {fastest['backend']} - {fastest['init_time']:.1f}ms")
print(f"🐌 最慢后端: {slowest['backend']} - {slowest['init_time']:.1f}ms")
if fastest['init_time'] > 0 and slowest['init_time'] > 0:
improvement = ((slowest['init_time'] - fastest['init_time']) / slowest['init_time']) * 100
print(f"💡 性能提升: {improvement:.1f}% (使用最快后端)")
print(f"\n📋 详细性能对比:")
for result in successful_results:
if result != fastest:
slowdown = ((result['init_time'] - fastest['init_time']) / fastest['init_time']) * 100
print(f" {result['backend']}: 比最快后端慢 {slowdown:.1f}% ({result['init_time']:.1f}ms vs {fastest['init_time']:.1f}ms)")
print(f"\n🎯 建议:")
if successful_results:
fastest = min(successful_results, key=lambda x: x['init_time'])
print(f"✅ 推荐使用 {fastest['backend']} 后端以获得最佳性能")
print(f"📝 配置建议: 在配置文件中设置 backend = {fastest['backend']}")
if fastest['init_time'] > 5000:
print(f"⚠️ 性能评级: 需要优化 (> 5秒)")
elif fastest['init_time'] > 2000:
print(f"⚠️ 性能评级: 一般 (2-5秒)")
else:
print(f"✅ 性能评级: 良好 (< 2秒)")
else:
print(f"❌ 所有后端测试失败,请检查相机连接")
print(f"\n{'='*80}")
print(f"测试完成")
print(f"{'='*80}")
if __name__ == "__main__":
import cv2 # 需要导入cv2用于相机操作
main()

View File

@ -0,0 +1,143 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
深入分析相机设备的行为
"""
import cv2
import numpy as np
import time
def analyze_camera_frame(frame, device_index):
"""
分析相机帧的特征
"""
print(f"\n=== 设备 {device_index} 帧分析 ===")
print(f"帧形状: {frame.shape}")
print(f"数据类型: {frame.dtype}")
print(f"数据范围: {frame.min()} - {frame.max()}")
# 检查是否是纯色帧(可能是虚拟设备)
unique_colors = len(np.unique(frame.reshape(-1, frame.shape[-1]), axis=0))
print(f"唯一颜色数量: {unique_colors}")
# 检查帧的统计信息
mean_values = np.mean(frame, axis=(0, 1))
std_values = np.std(frame, axis=(0, 1))
print(f"各通道均值: {mean_values}")
print(f"各通道标准差: {std_values}")
# 检查是否是静态帧
if np.all(std_values < 1.0):
print("⚠️ 这可能是一个静态/虚拟帧(标准差很小)")
# 检查是否是纯黑帧
if np.all(mean_values < 10):
print("⚠️ 这可能是一个黑色帧")
# 检查帧的变化
return frame
def test_camera_devices():
"""
测试多个相机设备并比较帧内容
"""
print("=== 相机设备详细分析 ===")
devices_to_test = [0, 1]
frames = {}
for device_index in devices_to_test:
print(f"\n--- 测试设备 {device_index} ---")
try:
cap = cv2.VideoCapture(device_index, cv2.CAP_DSHOW)
if cap.isOpened():
print(f"设备 {device_index}: 成功打开")
# 获取设备属性
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = cap.get(cv2.CAP_PROP_FPS)
fourcc = int(cap.get(cv2.CAP_PROP_FOURCC))
print(f"分辨率: {width}x{height}")
print(f"帧率: {fps}")
print(f"编码: {fourcc}")
# 读取多帧进行分析
frames_list = []
for i in range(5):
ret, frame = cap.read()
if ret and frame is not None:
frames_list.append(frame.copy())
if i == 0: # 只分析第一帧
frames[device_index] = analyze_camera_frame(frame, device_index)
else:
print(f"{i+1}帧读取失败")
break
# 检查帧间变化
if len(frames_list) > 1:
diff = cv2.absdiff(frames_list[0], frames_list[-1])
total_diff = np.sum(diff)
print(f"首末帧差异总和: {total_diff}")
if total_diff < 1000: # 阈值可调整
print("⚠️ 帧内容几乎没有变化,可能是虚拟设备")
else:
print("✓ 帧内容有变化,可能是真实相机")
else:
print(f"设备 {device_index}: 无法打开")
cap.release()
except Exception as e:
print(f"设备 {device_index} 测试异常: {e}")
# 比较不同设备的帧
if 0 in frames and 1 in frames:
print("\n=== 设备间帧比较 ===")
diff = cv2.absdiff(frames[0], frames[1])
total_diff = np.sum(diff)
print(f"设备0和设备1帧差异总和: {total_diff}")
if total_diff < 1000:
print("⚠️ 两个设备的帧几乎相同设备1可能是设备0的镜像或虚拟设备")
else:
print("✓ 两个设备的帧不同,可能是独立的相机")
def check_system_cameras():
"""
检查系统中可用的相机设备
"""
print("\n=== 系统相机设备检查 ===")
available_cameras = []
# 测试前10个设备索引
for i in range(10):
cap = cv2.VideoCapture(i, cv2.CAP_DSHOW)
if cap.isOpened():
ret, _ = cap.read()
if ret:
available_cameras.append(i)
print(f"设备 {i}: 可用")
else:
print(f"设备 {i}: 打开但无法读取")
else:
print(f"设备 {i}: 不可用")
cap.release()
# 避免测试太多设备
if len(available_cameras) >= 3:
break
print(f"\n发现 {len(available_cameras)} 个可用相机设备: {available_cameras}")
return available_cameras
if __name__ == "__main__":
check_system_cameras()
test_camera_devices()

View File

@ -0,0 +1,194 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
相机断开连接测试脚本
测试相机USB拔出时是否能正常检测设备断连并发送socket信息
"""
import time
import threading
import logging
from devices.camera_manager import CameraManager
from devices.utils.config_manager import ConfigManager
from unittest.mock import Mock
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class MockSocketIO:
"""模拟SocketIO用于测试"""
def __init__(self):
self.events = []
self.lock = threading.Lock()
def emit(self, event, data, namespace=None):
"""记录发送的事件"""
with self.lock:
self.events.append({
'event': event,
'data': data,
'namespace': namespace,
'timestamp': time.time()
})
logger.info(f"Socket事件: {event} -> {data} (namespace: {namespace})")
def get_events(self):
"""获取所有事件"""
with self.lock:
return self.events.copy()
def clear_events(self):
"""清空事件记录"""
with self.lock:
self.events.clear()
def test_camera_disconnect_detection():
"""
测试相机断开连接检测功能
"""
logger.info("="*60)
logger.info("开始测试相机断开连接检测功能")
logger.info("="*60)
# 创建模拟SocketIO
mock_socketio = MockSocketIO()
# 创建配置管理器
config_manager = ConfigManager()
# 创建相机管理器
camera_manager = CameraManager(mock_socketio, config_manager)
try:
# 1. 初始化相机
logger.info("\n步骤1: 初始化相机设备")
if not camera_manager.initialize():
logger.error("相机初始化失败,无法进行测试")
return False
logger.info(f"相机初始化成功 - 连接状态: {camera_manager.is_connected}")
# 2. 启动数据流
logger.info("\n步骤2: 启动相机数据流")
if not camera_manager.start_streaming():
logger.error("相机数据流启动失败")
return False
logger.info("相机数据流启动成功")
# 3. 等待一段时间让系统稳定
logger.info("\n步骤3: 等待系统稳定 (5秒)")
time.sleep(5)
# 清空之前的事件记录
mock_socketio.clear_events()
# 4. 提示用户拔出USB
logger.info("\n步骤4: 请拔出相机USB连接线")
logger.info("等待30秒来检测断开连接...")
start_time = time.time()
disconnect_detected = False
# 监控30秒
while time.time() - start_time < 30:
# 检查连接状态
if camera_manager.is_connected:
logger.debug(f"相机仍然连接中... (已等待 {time.time() - start_time:.1f}秒)")
else:
logger.info(f"检测到相机断开连接! (耗时 {time.time() - start_time:.1f}秒)")
disconnect_detected = True
break
time.sleep(1)
# 5. 分析结果
logger.info("\n步骤5: 分析测试结果")
if disconnect_detected:
logger.info("✓ 相机断开连接检测: 成功")
else:
logger.warning("✗ 相机断开连接检测: 失败 (30秒内未检测到断开)")
# 检查Socket事件
events = mock_socketio.get_events()
disconnect_events = [e for e in events if 'status' in str(e.get('data', {})) and 'disconnect' in str(e.get('data', {})).lower()]
if disconnect_events:
logger.info(f"✓ Socket断开通知: 成功 (发送了 {len(disconnect_events)} 个断开事件)")
for event in disconnect_events:
logger.info(f" - 事件: {event['event']}, 数据: {event['data']}")
else:
logger.warning("✗ Socket断开通知: 失败 (未发送断开事件)")
# 6. 测试重连机制
if disconnect_detected:
logger.info("\n步骤6: 测试重连机制")
logger.info("请重新插入相机USB连接线")
logger.info("等待30秒来检测重新连接...")
start_time = time.time()
reconnect_detected = False
while time.time() - start_time < 30:
if camera_manager.is_connected:
logger.info(f"检测到相机重新连接! (耗时 {time.time() - start_time:.1f}秒)")
reconnect_detected = True
break
time.sleep(1)
if reconnect_detected:
logger.info("✓ 相机重连检测: 成功")
else:
logger.warning("✗ 相机重连检测: 失败 (30秒内未检测到重连)")
# 7. 显示所有Socket事件
logger.info("\n步骤7: 所有Socket事件记录")
all_events = mock_socketio.get_events()
if all_events:
for i, event in enumerate(all_events, 1):
logger.info(f" {i}. 事件: {event['event']}, 数据: {event['data']}, 时间: {time.strftime('%H:%M:%S', time.localtime(event['timestamp']))}")
else:
logger.info(" 无Socket事件记录")
return disconnect_detected
except Exception as e:
logger.error(f"测试过程中发生异常: {e}")
return False
finally:
# 清理资源
try:
camera_manager.stop_streaming()
camera_manager.disconnect()
logger.info("测试资源清理完成")
except Exception as e:
logger.error(f"清理资源时发生异常: {e}")
def main():
"""
主函数
"""
logger.info("相机断开连接测试脚本")
logger.info("此脚本将测试相机USB拔出时的断连检测和Socket通知功能")
logger.info("")
# 运行测试
success = test_camera_disconnect_detection()
logger.info("\n" + "="*60)
if success:
logger.info("测试完成: 相机断开连接检测功能正常")
else:
logger.info("测试完成: 相机断开连接检测功能存在问题")
logger.info("="*60)
if __name__ == "__main__":
main()

212
backend/test_camera_full.py Normal file
View File

@ -0,0 +1,212 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
完整的相机断开连接测试脚本
模拟主程序的完整流程
"""
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
import time
import threading
import logging
from datetime import datetime
from devices.camera_manager import CameraManager
from devices.utils.config_manager import ConfigManager
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class MockSocketIO:
"""模拟SocketIO用于测试"""
def __init__(self):
self.events = []
self.lock = threading.Lock()
def emit(self, event, data, namespace=None):
"""记录发送的事件"""
with self.lock:
self.events.append({
'event': event,
'data': data,
'namespace': namespace,
'timestamp': time.time()
})
logger.info(f"Socket事件: {event} -> {data} (namespace: {namespace})")
def get_events(self):
"""获取所有事件"""
with self.lock:
return self.events.copy()
def clear_events(self):
"""清空事件记录"""
with self.lock:
self.events.clear()
class MockAppServer:
"""模拟主程序服务器"""
def __init__(self):
self.socketio = MockSocketIO()
self.logger = logger
self.device_managers = {}
def broadcast_device_status(self, device_name: str, is_connected: bool):
"""广播单个设备状态"""
if self.socketio:
try:
status_data = {
'device_type': device_name,
'status': is_connected,
'timestamp': datetime.now().isoformat()
}
self.socketio.emit('device_status', status_data, namespace='/devices')
self.logger.info(f'广播设备状态: {device_name} -> {"已连接" if is_connected else "未连接"}')
except Exception as e:
self.logger.error(f'广播设备状态失败: {e}')
def _on_device_status_change(self, device_name: str, is_connected: bool):
"""设备状态变化回调函数"""
self.logger.info(f'设备状态变化: {device_name} -> {"已连接" if is_connected else "未连接"}')
self.broadcast_device_status(device_name, is_connected)
def test_camera_disconnect_with_socket():
"""测试相机断开连接和Socket通知"""
logger.info("="*60)
logger.info("开始测试相机断开连接和Socket通知功能")
logger.info("="*60)
# 创建模拟服务器
app_server = MockAppServer()
try:
# 创建配置管理器
config_manager = ConfigManager()
# 创建相机管理器
camera_manager = CameraManager(app_server.socketio, config_manager)
app_server.device_managers['camera'] = camera_manager
# 添加状态变化回调(模拟主程序的回调注册)
camera_manager.add_status_change_callback(app_server._on_device_status_change)
# 1. 测试初始化
logger.info("\n步骤1: 初始化相机设备")
if camera_manager.initialize():
logger.info(f"✓ 相机初始化成功 - 连接状态: {camera_manager.is_connected}")
else:
logger.warning("✗ 相机初始化失败")
return False
# 清空初始化时的事件
app_server.socketio.clear_events()
# 2. 启动数据流(可选)
logger.info("\n步骤2: 启动相机数据流")
try:
if camera_manager.start_streaming(app_server.socketio):
logger.info("✓ 相机数据流启动成功")
time.sleep(2) # 让数据流稳定
else:
logger.warning("✗ 相机数据流启动失败")
except Exception as e:
logger.warning(f"数据流启动异常: {e}")
# 3. 监控连接状态变化
logger.info("\n步骤3: 监控连接状态变化 (30秒)")
logger.info("请在此期间拔出相机USB连接线来测试断开检测...")
start_time = time.time()
last_status = camera_manager.is_connected
disconnect_detected = False
while time.time() - start_time < 30:
current_status = camera_manager.is_connected
if current_status != last_status:
elapsed_time = time.time() - start_time
logger.info(f"检测到状态变化: {'连接' if current_status else '断开'} (耗时: {elapsed_time:.1f}秒)")
last_status = current_status
if not current_status:
logger.info("✓ 成功检测到相机断开!")
disconnect_detected = True
time.sleep(2) # 等待事件处理完成
break
time.sleep(0.5)
# 4. 检查Socket事件
logger.info("\n步骤4: 检查Socket事件")
events = app_server.socketio.get_events()
if events:
logger.info(f"✓ 共记录到 {len(events)} 个Socket事件:")
disconnect_events = 0
for i, event in enumerate(events, 1):
logger.info(f" {i}. 事件: {event['event']}, 数据: {event['data']}, 命名空间: {event['namespace']}")
if event['event'] == 'device_status' and event['data'].get('status') == False:
disconnect_events += 1
if disconnect_events > 0:
logger.info(f"✓ 检测到 {disconnect_events} 个设备断开事件")
else:
logger.warning("✗ 未检测到设备断开事件")
else:
logger.warning("✗ 未记录到任何Socket事件")
# 5. 测试结果总结
logger.info("\n步骤5: 测试结果总结")
if disconnect_detected:
logger.info("✓ 硬件断开检测: 成功")
else:
logger.warning("✗ 硬件断开检测: 失败 (30秒内未检测到断开)")
if events and any(e['event'] == 'device_status' and e['data'].get('status') == False for e in events):
logger.info("✓ Socket断开通知: 成功")
else:
logger.warning("✗ Socket断开通知: 失败")
return disconnect_detected and len(events) > 0
except Exception as e:
logger.error(f"测试过程中发生异常: {e}")
import traceback
traceback.print_exc()
return False
finally:
# 清理资源
try:
if 'camera_manager' in locals():
camera_manager.stop_streaming()
camera_manager.disconnect()
logger.info("测试资源清理完成")
except Exception as e:
logger.error(f"清理资源时发生异常: {e}")
def main():
"""主函数"""
logger.info("相机断开连接完整测试脚本")
logger.info("此脚本将模拟主程序的完整流程测试相机USB拔出时的断连检测和Socket通知功能")
success = test_camera_disconnect_with_socket()
logger.info("\n" + "="*60)
if success:
logger.info("测试完成: 相机断开连接检测和Socket通知功能正常")
else:
logger.info("测试完成: 相机断开连接检测和Socket通知功能存在问题")
logger.info("="*60)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,217 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
相机启动性能测试脚本
"""
import sys
import os
import time
import logging
from typing import Dict, Any
# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from devices.utils.config_manager import ConfigManager
from devices.camera_manager import CameraManager
# 设置日志级别为DEBUG以查看详细信息
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
def test_camera_startup_performance():
"""
测试相机启动性能
"""
print("=" * 60)
print("相机启动性能测试")
print("=" * 60)
try:
# 初始化配置管理器
print("\n1. 初始化配置管理器...")
config_start = time.time()
config_manager = ConfigManager()
config_time = (time.time() - config_start) * 1000
print(f"配置管理器初始化完成 (耗时: {config_time:.1f}ms)")
# 创建相机管理器
print("\n2. 创建相机管理器...")
manager_start = time.time()
camera_manager = CameraManager(None, config_manager)
manager_time = (time.time() - manager_start) * 1000
print(f"相机管理器创建完成 (耗时: {manager_time:.1f}ms)")
# 测试多次初始化以获得平均性能
print("\n3. 执行相机初始化性能测试...")
test_rounds = 3
init_times = []
for i in range(test_rounds):
print(f"\n--- 第 {i+1} 轮测试 ---")
# 如果之前已连接,先断开
if camera_manager.is_connected:
disconnect_start = time.time()
camera_manager.disconnect()
disconnect_time = (time.time() - disconnect_start) * 1000
print(f"断开连接耗时: {disconnect_time:.1f}ms")
time.sleep(0.5) # 等待设备完全断开
# 执行初始化
init_start = time.time()
success = camera_manager.initialize()
init_time = (time.time() - init_start) * 1000
if success:
print(f"✓ 初始化成功 (总耗时: {init_time:.1f}ms)")
init_times.append(init_time)
# 测试校准性能
calibrate_start = time.time()
calibrate_success = camera_manager.calibrate()
calibrate_time = (time.time() - calibrate_start) * 1000
if calibrate_success:
print(f"✓ 校准成功 (耗时: {calibrate_time:.1f}ms)")
else:
print(f"✗ 校准失败 (耗时: {calibrate_time:.1f}ms)")
# 测试第一帧获取时间
if camera_manager.cap:
first_frame_start = time.time()
ret, frame = camera_manager.cap.read()
first_frame_time = (time.time() - first_frame_start) * 1000
if ret and frame is not None:
print(f"✓ 首帧获取成功 (耗时: {first_frame_time:.1f}ms, 帧大小: {frame.shape})")
del frame # 释放内存
else:
print(f"✗ 首帧获取失败 (耗时: {first_frame_time:.1f}ms)")
else:
print(f"✗ 初始化失败 (耗时: {init_time:.1f}ms)")
time.sleep(1) # 测试间隔
# 性能统计
print("\n" + "=" * 60)
print("性能统计结果")
print("=" * 60)
if init_times:
avg_init_time = sum(init_times) / len(init_times)
min_init_time = min(init_times)
max_init_time = max(init_times)
print(f"初始化性能统计 ({len(init_times)} 次成功测试):")
print(f" 平均耗时: {avg_init_time:.1f}ms")
print(f" 最快耗时: {min_init_time:.1f}ms")
print(f" 最慢耗时: {max_init_time:.1f}ms")
# 性能评估
if avg_init_time < 1000: # 1秒以内
print(f" 性能评级: 优秀 ⭐⭐⭐")
elif avg_init_time < 3000: # 3秒以内
print(f" 性能评级: 良好 ⭐⭐")
elif avg_init_time < 5000: # 5秒以内
print(f" 性能评级: 一般 ⭐")
else:
print(f" 性能评级: 需要优化 ❌")
else:
print("❌ 所有初始化测试都失败了")
# 获取设备信息
if camera_manager.is_connected:
print("\n设备信息:")
device_info = camera_manager.get_device_info()
for key, value in device_info.items():
print(f" {key}: {value}")
# 清理资源
print("\n4. 清理资源...")
cleanup_start = time.time()
camera_manager.cleanup()
cleanup_time = (time.time() - cleanup_start) * 1000
print(f"资源清理完成 (耗时: {cleanup_time:.1f}ms)")
except Exception as e:
print(f"\n❌ 测试过程中发生错误: {e}")
import traceback
traceback.print_exc()
def test_streaming_startup():
"""
测试流媒体启动性能
"""
print("\n" + "=" * 60)
print("流媒体启动性能测试")
print("=" * 60)
try:
config_manager = ConfigManager()
camera_manager = CameraManager(None, config_manager)
# 初始化相机
print("\n1. 初始化相机...")
if not camera_manager.initialize():
print("❌ 相机初始化失败,无法进行流媒体测试")
return
# 测试流媒体启动
print("\n2. 启动流媒体...")
streaming_start = time.time()
streaming_success = camera_manager.start_streaming()
streaming_time = (time.time() - streaming_start) * 1000
if streaming_success:
print(f"✓ 流媒体启动成功 (耗时: {streaming_time:.1f}ms)")
# 等待几秒钟收集帧数据
print("\n3. 收集性能数据...")
time.sleep(3)
# 获取统计信息
stats = camera_manager.get_stats()
print(f"\n流媒体性能统计:")
for key, value in stats.items():
print(f" {key}: {value}")
# 停止流媒体
print("\n4. 停止流媒体...")
stop_start = time.time()
camera_manager.stop_streaming()
stop_time = (time.time() - stop_start) * 1000
print(f"✓ 流媒体停止完成 (耗时: {stop_time:.1f}ms)")
else:
print(f"❌ 流媒体启动失败 (耗时: {streaming_time:.1f}ms)")
# 清理
camera_manager.cleanup()
except Exception as e:
print(f"\n❌ 流媒体测试过程中发生错误: {e}")
import traceback
traceback.print_exc()
def main():
"""
主函数
"""
print("相机性能测试工具")
print("测试目标优化相机启动时间目标从10+秒降低到3秒以内")
# 执行基本启动性能测试
test_camera_startup_performance()
# 执行流媒体启动性能测试
test_streaming_startup()
print("\n" + "=" * 60)
print("测试完成!")
print("=" * 60)
if __name__ == '__main__':
main()

View File

@ -0,0 +1,148 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
简化的相机断开连接测试脚本
"""
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
import time
import threading
import logging
from devices.camera_manager import CameraManager
from devices.utils.config_manager import ConfigManager
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class MockSocketIO:
"""模拟SocketIO用于测试"""
def __init__(self):
self.events = []
self.lock = threading.Lock()
def emit(self, event, data, namespace=None):
"""记录发送的事件"""
with self.lock:
self.events.append({
'event': event,
'data': data,
'namespace': namespace,
'timestamp': time.time()
})
logger.info(f"Socket事件: {event} -> {data} (namespace: {namespace})")
def get_events(self):
"""获取所有事件"""
with self.lock:
return self.events.copy()
def test_camera_connection():
"""测试相机连接和断开检测"""
logger.info("="*60)
logger.info("开始测试相机连接和断开检测功能")
logger.info("="*60)
# 创建模拟SocketIO
mock_socketio = MockSocketIO()
try:
# 创建配置管理器
config_manager = ConfigManager()
# 创建相机管理器
camera_manager = CameraManager(mock_socketio, config_manager)
# 添加状态变化回调
def status_callback(device_name, is_connected):
logger.info(f"状态回调: {device_name} -> {'连接' if is_connected else '断开'}")
camera_manager.add_status_change_callback(status_callback)
# 1. 测试初始化
logger.info("\n步骤1: 初始化相机设备")
if camera_manager.initialize():
logger.info(f"✓ 相机初始化成功 - 连接状态: {camera_manager.is_connected}")
else:
logger.warning("✗ 相机初始化失败")
return False
# 2. 测试硬件连接检查
logger.info("\n步骤2: 测试硬件连接检查")
hardware_connected = camera_manager.check_hardware_connection()
logger.info(f"硬件连接状态: {hardware_connected}")
# 3. 启动连接监控
logger.info("\n步骤3: 启动连接监控")
camera_manager._start_connection_monitor()
logger.info("连接监控已启动")
# 4. 监控连接状态变化
logger.info("\n步骤4: 监控连接状态 (30秒)")
logger.info("请在此期间拔出相机USB连接线来测试断开检测...")
start_time = time.time()
last_status = camera_manager.is_connected
while time.time() - start_time < 30:
current_status = camera_manager.is_connected
if current_status != last_status:
logger.info(f"检测到状态变化: {'连接' if current_status else '断开'} (耗时: {time.time() - start_time:.1f}秒)")
last_status = current_status
if not current_status:
logger.info("✓ 成功检测到相机断开!")
break
time.sleep(1)
# 5. 检查Socket事件
logger.info("\n步骤5: 检查Socket事件")
events = mock_socketio.get_events()
if events:
logger.info(f"共记录到 {len(events)} 个Socket事件:")
for i, event in enumerate(events, 1):
logger.info(f" {i}. {event['event']} -> {event['data']}")
else:
logger.info("未记录到Socket事件")
return True
except Exception as e:
logger.error(f"测试过程中发生异常: {e}")
return False
finally:
# 清理资源
try:
if 'camera_manager' in locals():
camera_manager._stop_connection_monitor()
camera_manager.disconnect()
logger.info("测试资源清理完成")
except Exception as e:
logger.error(f"清理资源时发生异常: {e}")
def main():
"""主函数"""
logger.info("相机断开连接测试脚本")
success = test_camera_connection()
logger.info("\n" + "="*60)
if success:
logger.info("测试完成: 相机断开连接检测功能测试完成")
else:
logger.info("测试完成: 相机断开连接检测功能存在问题")
logger.info("="*60)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,305 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
OpenCV后端性能测试脚本
测试不同OpenCV后端DirectShow vs MSMF的性能差异
"""
import sys
import os
import time
import logging
import cv2
# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_opencv_backend(backend_name, backend_id, device_index=0, width=1280, height=720, fps=30):
"""
测试指定OpenCV后端的性能
Args:
backend_name: 后端名称
backend_id: 后端ID
device_index: 设备索引
width: 宽度
height: 高度
fps: 帧率
Returns:
dict: 性能数据
"""
print(f"\n{'='*70}")
print(f"测试 {backend_name} 后端 (ID: {backend_id})")
print(f"分辨率: {width}x{height}, FPS: {fps}")
print(f"{'='*70}")
result = {
'backend_name': backend_name,
'backend_id': backend_id,
'success': False,
'init_time': -1,
'config_time': -1,
'first_frame_time': -1,
'total_time': -1,
'actual_resolution': 'N/A',
'error': None
}
cap = None
try:
# 1. 测试相机初始化时间
print(f"📷 初始化相机 (后端: {backend_name})...")
init_start = time.time()
# 创建VideoCapture对象并指定后端
cap = cv2.VideoCapture(device_index, backend_id)
if not cap.isOpened():
print(f"❌ 无法打开相机 (后端: {backend_name})")
result['error'] = f"无法打开相机 (后端: {backend_name})"
return result
init_time = (time.time() - init_start) * 1000
result['init_time'] = init_time
print(f"✅ 相机初始化成功: {init_time:.1f}ms")
# 2. 测试配置时间
print(f"⚙️ 配置相机参数...")
config_start = time.time()
# 设置分辨率和帧率
cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
cap.set(cv2.CAP_PROP_FPS, fps)
# 设置缓冲区大小
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
# 性能优化设置
try:
cap.set(cv2.CAP_PROP_AUTO_EXPOSURE, 0.25) # 手动曝光
cap.set(cv2.CAP_PROP_AUTO_WB, 0) # 禁用自动白平衡
cap.set(cv2.CAP_PROP_EXPOSURE, -6) # 设置曝光值
except Exception as e:
print(f"⚠️ 性能优化设置警告: {e}")
config_time = (time.time() - config_start) * 1000
result['config_time'] = config_time
print(f"✅ 配置完成: {config_time:.1f}ms")
# 3. 获取实际分辨率
actual_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
actual_fps = cap.get(cv2.CAP_PROP_FPS)
result['actual_resolution'] = f"{actual_width}x{actual_height}@{actual_fps:.1f}fps"
print(f"🎯 实际参数: {actual_width}x{actual_height}, FPS: {actual_fps:.1f}")
# 4. 测试首帧获取时间
print(f"🖼️ 获取首帧...")
frame_start = time.time()
ret, frame = cap.read()
if ret and frame is not None:
first_frame_time = (time.time() - frame_start) * 1000
result['first_frame_time'] = first_frame_time
print(f"✅ 首帧获取成功: {first_frame_time:.1f}ms, 帧大小: {frame.shape}")
else:
print(f"❌ 首帧获取失败")
result['error'] = "首帧获取失败"
return result
# 5. 计算总时间
total_time = init_time + config_time + first_frame_time
result['total_time'] = total_time
result['success'] = True
print(f"📊 总耗时: {total_time:.1f}ms ({total_time/1000:.2f}秒)")
# 6. 测试连续帧获取性能
print(f"🎬 测试连续帧获取性能...")
frame_times = []
test_frames = 10
for i in range(test_frames):
frame_start = time.time()
ret, frame = cap.read()
if ret:
frame_time = (time.time() - frame_start) * 1000
frame_times.append(frame_time)
else:
break
if frame_times:
avg_frame_time = sum(frame_times) / len(frame_times)
max_frame_time = max(frame_times)
min_frame_time = min(frame_times)
print(f"📈 连续帧性能: 平均 {avg_frame_time:.1f}ms, 最快 {min_frame_time:.1f}ms, 最慢 {max_frame_time:.1f}ms")
result['avg_frame_time'] = avg_frame_time
result['max_frame_time'] = max_frame_time
result['min_frame_time'] = min_frame_time
return result
except Exception as e:
print(f"❌ 测试异常: {e}")
result['error'] = str(e)
return result
finally:
if cap:
cap.release()
print(f"🧹 相机资源已释放")
def get_available_backends():
"""
获取可用的OpenCV后端
Returns:
list: 可用后端列表
"""
backends = [
('DirectShow', cv2.CAP_DSHOW),
('MSMF', cv2.CAP_MSMF),
('V4L2', cv2.CAP_V4L2),
('GStreamer', cv2.CAP_GSTREAMER),
('Any', cv2.CAP_ANY)
]
available_backends = []
for name, backend_id in backends:
try:
# 尝试创建VideoCapture对象
cap = cv2.VideoCapture(0, backend_id)
if cap.isOpened():
available_backends.append((name, backend_id))
cap.release()
except Exception:
pass
return available_backends
def main():
"""
主测试函数
"""
print("🚀 OpenCV后端性能测试")
print(f"OpenCV版本: {cv2.__version__}")
# 获取可用后端
print("\n🔍 检测可用的OpenCV后端...")
available_backends = get_available_backends()
if not available_backends:
print("❌ 未找到可用的相机后端")
return
print(f"✅ 找到 {len(available_backends)} 个可用后端:")
for name, backend_id in available_backends:
print(f" - {name} (ID: {backend_id})")
# 测试参数
test_params = {
'device_index': 0,
'width': 1280,
'height': 720,
'fps': 30
}
print(f"\n📋 测试参数: {test_params['width']}x{test_params['height']}@{test_params['fps']}fps")
# 执行测试
results = []
for backend_name, backend_id in available_backends:
result = test_opencv_backend(
backend_name,
backend_id,
**test_params
)
results.append(result)
# 等待一下,避免设备冲突
time.sleep(2)
# 输出汇总结果
print(f"\n{'='*90}")
print("📈 OpenCV后端性能测试汇总")
print(f"{'='*90}")
print(f"{'后端':<12} {'状态':<8} {'初始化':<10} {'配置':<10} {'首帧':<10} {'总计':<10} {'实际分辨率':<20}")
print("-" * 90)
successful_results = []
for result in results:
status = "✅成功" if result['success'] else "❌失败"
init_time = f"{result['init_time']:.1f}ms" if result['init_time'] > 0 else "N/A"
config_time = f"{result['config_time']:.1f}ms" if result['config_time'] > 0 else "N/A"
frame_time = f"{result['first_frame_time']:.1f}ms" if result['first_frame_time'] > 0 else "N/A"
total_time = f"{result['total_time']:.1f}ms" if result['total_time'] > 0 else "N/A"
print(f"{result['backend_name']:<12} {status:<8} {init_time:<10} {config_time:<10} {frame_time:<10} {total_time:<10} {result['actual_resolution']:<20}")
if result['success']:
successful_results.append(result)
# 性能分析
if len(successful_results) >= 2:
print(f"\n📊 性能分析:")
# 找到最快和最慢的后端
fastest = min(successful_results, key=lambda x: x['total_time'])
slowest = max(successful_results, key=lambda x: x['total_time'])
print(f"🏆 最快后端: {fastest['backend_name']} - {fastest['total_time']:.1f}ms")
print(f"🐌 最慢后端: {slowest['backend_name']} - {slowest['total_time']:.1f}ms")
if slowest['total_time'] > fastest['total_time']:
improvement = ((slowest['total_time'] - fastest['total_time']) / slowest['total_time']) * 100
print(f"💡 性能提升: {improvement:.1f}% (使用最快后端)")
# 详细对比
print(f"\n📋 详细性能对比:")
for result in successful_results:
if result != fastest:
if result['total_time'] > fastest['total_time']:
slower = ((result['total_time'] - fastest['total_time']) / fastest['total_time']) * 100
print(f" {result['backend_name']}: 比最快后端慢 {slower:.1f}% ({result['total_time']:.1f}ms vs {fastest['total_time']:.1f}ms)")
elif len(successful_results) == 1:
result = successful_results[0]
print(f"\n📊 只有一个后端测试成功: {result['backend_name']} - {result['total_time']:.1f}ms")
# 推荐建议
print(f"\n🎯 建议:")
if successful_results:
fastest = min(successful_results, key=lambda x: x['total_time'])
print(f"✅ 推荐使用 {fastest['backend_name']} 后端以获得最佳性能")
print(f"📝 配置建议: 在相机初始化时指定后端 cv2.VideoCapture(device_index, cv2.CAP_{fastest['backend_name'].upper()})")
if fastest['total_time'] < 3000:
print(f"🚀 性能评级: 优秀 (< 3秒)")
elif fastest['total_time'] < 5000:
print(f"⚡ 性能评级: 良好 (< 5秒)")
else:
print(f"⚠️ 性能评级: 需要优化 (> 5秒)")
else:
print(f"❌ 所有后端测试都失败了,请检查相机连接和驱动")
print(f"\n{'='*90}")
print("测试完成")
print(f"{'='*90}")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,113 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试OpenCV VideoCapture的行为
验证当设备索引不存在时VideoCapture的表现
"""
import cv2
import time
def test_video_capture_behavior():
"""
测试不同设备索引的VideoCapture行为
"""
print("=== OpenCV VideoCapture 行为测试 ===")
print(f"OpenCV版本: {cv2.__version__}")
print()
# 测试不同的设备索引
test_indices = [0, 1, 2, 3, -1]
backends = [cv2.CAP_DSHOW, cv2.CAP_MSMF, cv2.CAP_ANY]
backend_names = ['CAP_DSHOW', 'CAP_MSMF', 'CAP_ANY']
for device_index in test_indices:
print(f"\n--- 测试设备索引 {device_index} ---")
for backend, backend_name in zip(backends, backend_names):
print(f"\n后端: {backend_name}")
try:
start_time = time.time()
cap = cv2.VideoCapture(device_index, backend)
open_time = (time.time() - start_time) * 1000
print(f" VideoCapture创建: 成功 (耗时: {open_time:.1f}ms)")
print(f" isOpened(): {cap.isOpened()}")
if cap.isOpened():
# 尝试读取帧
start_time = time.time()
ret, frame = cap.read()
read_time = (time.time() - start_time) * 1000
print(f" read()返回值: ret={ret}")
if ret and frame is not None:
print(f" 帧形状: {frame.shape}")
print(f" 读取耗时: {read_time:.1f}ms")
else:
print(f" 读取失败 (耗时: {read_time:.1f}ms)")
# 获取一些属性
try:
width = cap.get(cv2.CAP_PROP_FRAME_WIDTH)
height = cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
fps = cap.get(cv2.CAP_PROP_FPS)
print(f" 分辨率: {int(width)}x{int(height)}")
print(f" 帧率: {fps}")
except Exception as e:
print(f" 获取属性失败: {e}")
else:
print(" 相机未打开")
cap.release()
except Exception as e:
print(f" 异常: {e}")
print("\n=== 测试完成 ===")
def test_specific_case():
"""
专门测试device_index=1的情况
"""
print("\n=== 专门测试 device_index=1 ===")
try:
# 使用DSHOW后端Windows默认
cap = cv2.VideoCapture(1, cv2.CAP_DSHOW)
print(f"VideoCapture(1, CAP_DSHOW) 创建成功")
print(f"isOpened(): {cap.isOpened()}")
if cap.isOpened():
print("相机显示为已打开,但这可能是虚假的")
# 尝试多次读取
for i in range(3):
print(f"\n{i+1}次读取:")
start_time = time.time()
ret, frame = cap.read()
read_time = (time.time() - start_time) * 1000
print(f" ret: {ret}")
print(f" frame is None: {frame is None}")
print(f" 耗时: {read_time:.1f}ms")
if ret and frame is not None:
print(f" 帧形状: {frame.shape}")
print(f" 帧数据类型: {frame.dtype}")
print(f" 帧数据范围: {frame.min()} - {frame.max()}")
else:
print(" 读取失败或帧为空")
break
else:
print("相机未打开")
cap.release()
except Exception as e:
print(f"异常: {e}")
if __name__ == "__main__":
test_video_capture_behavior()
test_specific_case()

View File

@ -0,0 +1,98 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
设备重连机制测试脚本
测试设备断开后的自动重连功能
"""
import time
import threading
from devices.camera_manager import CameraManager
from devices.imu_manager import IMUManager
from devices.femtobolt_manager import FemtoBoltManager
from devices.pressure_manager import PressureManager
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
class MockSocketIO:
"""模拟SocketIO用于测试"""
def emit(self, event, data):
print(f"[SocketIO] 发送事件: {event}, 数据: {data}")
def test_device_reconnection(device_manager, device_name):
"""测试设备重连机制"""
print(f"\n=== 测试 {device_name} 重连机制 ===")
# 初始化设备
print(f"1. 初始化 {device_name} 设备...")
success = device_manager.initialize()
print(f" 初始化结果: {'成功' if success else '失败'}")
if success:
print(f" 设备连接状态: {'已连接' if device_manager.is_connected else '未连接'}")
# 等待一段时间让连接稳定
print("2. 等待连接稳定...")
time.sleep(3)
# 模拟设备断开
print("3. 模拟设备断开连接...")
device_manager.disconnect()
print(f" 断开后连接状态: {'已连接' if device_manager.is_connected else '未连接'}")
# 等待一段时间
print("4. 等待重连机制触发...")
time.sleep(5)
# 尝试重新连接
print("5. 尝试重新连接...")
reconnect_success = device_manager.initialize()
print(f" 重连结果: {'成功' if reconnect_success else '失败'}")
print(f" 重连后连接状态: {'已连接' if device_manager.is_connected else '未连接'}")
# 清理
device_manager.disconnect()
print(f"=== {device_name} 重连测试完成 ===\n")
return success
def main():
"""主测试函数"""
print("开始设备重连机制测试...")
# 创建模拟SocketIO
mock_socketio = MockSocketIO()
# 测试相机重连
print("\n测试相机重连机制...")
camera_manager = CameraManager(mock_socketio)
test_device_reconnection(camera_manager, "相机")
# 测试IMU重连
print("\n测试IMU重连机制...")
imu_manager = IMUManager(mock_socketio)
test_device_reconnection(imu_manager, "IMU")
# 测试FemtoBolt重连
print("\n测试FemtoBolt重连机制...")
femtobolt_manager = FemtoBoltManager(mock_socketio)
test_device_reconnection(femtobolt_manager, "FemtoBolt")
# 测试压力传感器重连
print("\n测试压力传感器重连机制...")
pressure_manager = PressureManager(mock_socketio)
test_device_reconnection(pressure_manager, "压力传感器")
print("\n所有设备重连测试完成!")
print("\n注意事项:")
print("1. 某些设备可能需要物理连接才能成功初始化")
print("2. 重连机制的效果取决于设备的实际可用性")
print("3. 观察日志中的连接监控线程启动和停止信息")
if __name__ == "__main__":
main()

View File

@ -0,0 +1,214 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
分辨率性能测试脚本
测试不同分辨率下相机配置的性能差异
"""
import sys
import os
import time
import logging
# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from devices.camera_manager import CameraManager
from devices.utils.config_manager import ConfigManager
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_resolution_performance(width, height, test_name):
"""
测试指定分辨率的性能
Args:
width: 宽度
height: 高度
test_name: 测试名称
Returns:
dict: 性能数据
"""
print(f"\n{'='*60}")
print(f"测试 {test_name}: {width}x{height}")
print(f"{'='*60}")
# 创建配置管理器并设置分辨率
config_manager = ConfigManager()
# 获取原始配置
original_config = config_manager.get_device_config('camera')
# 临时设置测试分辨率
test_config = {
'width': width,
'height': height
}
config_manager.set_camera_config(test_config)
try:
# 创建相机管理器
camera = CameraManager(None, config_manager)
# 测试初始化性能
start_time = time.time()
success = camera.initialize()
total_time = (time.time() - start_time) * 1000
if success:
print(f"✅ 初始化成功")
print(f"📊 总耗时: {total_time:.1f}ms ({total_time/1000:.1f}秒)")
# 获取实际分辨率
if camera.cap:
actual_width = int(camera.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_height = int(camera.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
print(f"🎯 实际分辨率: {actual_width}x{actual_height}")
# 测试首帧获取
frame_start = time.time()
ret, frame = camera.cap.read() if camera.cap else (False, None)
frame_time = (time.time() - frame_start) * 1000
if ret and frame is not None:
print(f"🖼️ 首帧获取: {frame_time:.1f}ms, 帧大小: {frame.shape}")
else:
print(f"❌ 首帧获取失败")
frame_time = -1
# 清理资源
camera.cleanup()
return {
'resolution': f"{width}x{height}",
'success': True,
'total_time': total_time,
'frame_time': frame_time,
'actual_resolution': f"{actual_width}x{actual_height}" if camera.cap else "未知"
}
else:
print(f"❌ 初始化失败")
return {
'resolution': f"{width}x{height}",
'success': False,
'total_time': total_time,
'frame_time': -1,
'actual_resolution': "失败"
}
except Exception as e:
print(f"❌ 测试异常: {e}")
return {
'resolution': f"{width}x{height}",
'success': False,
'total_time': -1,
'frame_time': -1,
'actual_resolution': "异常",
'error': str(e)
}
finally:
# 恢复原始配置
try:
restore_config = {
'width': original_config['width'],
'height': original_config['height']
}
config_manager.set_camera_config(restore_config)
except Exception as e:
print(f"⚠️ 恢复配置失败: {e}")
def main():
"""
主测试函数
"""
print("🚀 开始分辨率性能测试")
# 测试不同分辨率
test_cases = [
(1280, 720, "当前分辨率"),
(640, 480, "标准VGA"),
(320, 240, "QVGA小分辨率"),
(160, 120, "极小分辨率")
]
results = []
for width, height, name in test_cases:
result = test_resolution_performance(width, height, name)
results.append(result)
# 等待一下,避免设备冲突
time.sleep(1)
# 输出汇总结果
print(f"\n{'='*80}")
print("📈 性能测试汇总")
print(f"{'='*80}")
print(f"{'分辨率':<15} {'状态':<8} {'初始化耗时':<12} {'首帧耗时':<10} {'实际分辨率':<15}")
print("-" * 80)
successful_results = []
for result in results:
status = "✅成功" if result['success'] else "❌失败"
init_time = f"{result['total_time']:.1f}ms" if result['total_time'] > 0 else "N/A"
frame_time = f"{result['frame_time']:.1f}ms" if result['frame_time'] > 0 else "N/A"
print(f"{result['resolution']:<15} {status:<8} {init_time:<12} {frame_time:<10} {result['actual_resolution']:<15}")
if result['success'] and result['total_time'] > 0:
successful_results.append(result)
# 性能分析
if len(successful_results) >= 2:
print(f"\n📊 性能分析:")
# 找到最快和最慢的
fastest = min(successful_results, key=lambda x: x['total_time'])
slowest = max(successful_results, key=lambda x: x['total_time'])
print(f"🏆 最快配置: {fastest['resolution']} - {fastest['total_time']:.1f}ms")
print(f"🐌 最慢配置: {slowest['resolution']} - {slowest['total_time']:.1f}ms")
if slowest['total_time'] > fastest['total_time']:
improvement = ((slowest['total_time'] - fastest['total_time']) / slowest['total_time']) * 100
print(f"💡 性能提升: {improvement:.1f}% (使用最小分辨率)")
# 基准对比
baseline = next((r for r in successful_results if "1280x720" in r['resolution']), None)
if baseline:
print(f"\n📋 相对于当前分辨率(1280x720)的性能对比:")
for result in successful_results:
if result != baseline:
if result['total_time'] < baseline['total_time']:
improvement = ((baseline['total_time'] - result['total_time']) / baseline['total_time']) * 100
print(f" {result['resolution']}: 快 {improvement:.1f}% ({result['total_time']:.1f}ms vs {baseline['total_time']:.1f}ms)")
else:
degradation = ((result['total_time'] - baseline['total_time']) / baseline['total_time']) * 100
print(f" {result['resolution']}: 慢 {degradation:.1f}% ({result['total_time']:.1f}ms vs {baseline['total_time']:.1f}ms)")
print(f"\n🎯 建议:")
if successful_results:
fastest = min(successful_results, key=lambda x: x['total_time'])
if fastest['total_time'] < 3000: # 小于3秒
print(f"✅ 推荐使用 {fastest['resolution']} 以获得最佳性能")
else:
print(f"⚠️ 即使最快的分辨率 {fastest['resolution']} 仍需 {fastest['total_time']:.1f}ms")
print(f" 建议考虑其他优化方案(如更换相机后端)")
else:
print(f"❌ 所有测试都失败了,请检查相机连接")
print(f"\n{'='*80}")
print("测试完成")
print(f"{'='*80}")
if __name__ == "__main__":
import cv2 # 在这里导入cv2避免在函数中导入
main()

View File

@ -11,7 +11,9 @@
<!-- <el-icon class="back-icon" @click="handleBack"><ArrowLeft /></el-icon> -->
<span class="page-title">实时检测</span>
</div>
<div style="padding-left: 10px;">检测中</div>
<div style="padding-left: 10px;">
<span :style="{ color: isConnected ? '#00C851' : '#FF4444' }">{{ isConnected ? '检测中' : '服务已断开!' }}</span>
</div>
<img src="@/assets/sz.png" alt="" title="编辑相机参数" v-if="isConnected == true"
style="margin-left: 20px;cursor: pointer; width: 24px;height: 24px;"
@click="cameraUpdate">
@ -26,7 +28,7 @@
class="start-title-btn" style="background-image: linear-gradient(to right, rgb(236, 50, 166), rgb(160, 5, 216));
--el-button-border-color: #409EFF;border:0px;
--el-button-border-color: transparent;width: 110px;height: 30px;font-size: 18px;">
{{ isConnected ? '开始录像' : '录像中...' }}
开始录像
</el-button>
<!-- handleStartStop -->
<el-button v-if="isRecording" @click="handleStartStop" type="primary" class="start-title-btn" style="background-image: linear-gradient(to right, rgb(236, 50, 166), rgb(160, 5, 216));