ThbTech/BodyBalanceEvaluation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of http://121.37.111.42:3000/ThbTech/BodyBalanceEvaluation into main

Browse Source
This commit is contained in:
jingna 2025-08-06 16:08:51 +08:00
commit bc3ffa474d
4 changed files with 387 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.gitignore vendored
View File

@ -11,6 +11,7 @@ frontend/src/renderer/node_modules/
# Python 缓存文件
__pycache__/
backend/__pycache__/
*.pyc
*.pyo
*.pyd
@ -44,4 +45,8 @@ build/
# 临时文件
*.tmp
*.temp
*.temp
backend/__pycache__/app.cpython-311.pyc
backend/__pycache__/app.cpython-311.pyc
backend/__pycache__/database.cpython-311.pyc
backend/__pycache__/app.cpython-311.pyc

72
backend/app.py
View File

@ -107,7 +107,10 @@ def init_app():
device_manager = DeviceManager(db_manager)
device_manager.set_socketio(socketio) # 设置WebSocket连接
# 自动启动推流以填充帧缓存
if device_manager.device_status['camera']:
streaming_result = device_manager.start_streaming()
logger.info(f'自动启动推流结果: {streaming_result}')
# 初始化视频流管理器
video_stream_manager = VideoStreamManager(socketio)
@ -138,6 +141,31 @@ def api_health_check():
'version': '1.0.0'
})
@app.route('/api/frame-cache/status', methods=['GET'])
def get_frame_cache_status():
"""获取帧缓存状态"""
try:
if device_manager:
cache_info = device_manager.get_frame_cache_info()
return jsonify({
'success': True,
'data': cache_info,
'timestamp': datetime.now().isoformat()
})
else:
return jsonify({
'success': False,
'error': '设备管理器未初始化',
'timestamp': datetime.now().isoformat()
}), 500
except Exception as e:
logger.error(f'获取帧缓存状态失败: {e}')
return jsonify({
'success': False,
'error': str(e),
'timestamp': datetime.now().isoformat()
}), 500
# ==================== 认证API ====================
@app.route('/api/auth/login', methods=['POST'])
@ -594,6 +622,7 @@ def calibrate_imu():
# ==================== 视频推流API ====================
@app.route('/api/streaming/start', methods=['POST'])
def start_video_streaming():
"""启动视频推流"""
@ -694,21 +723,23 @@ def stop_detection(session_id):
# video_data = data.get('videoData') if data else None
video_data = data['videoData']
mime_type = data.get('mimeType', 'video/webm;codecs=vp9') # 默认webm格式
import base64
# 验证base64视频数据格式
if not video_data.startswith('data:video/'):
return jsonify({
'success': False,
'message': '无效的视频数据格式'
}), 400
# try:
# header, encoded = video_data.split(',', 1)
# video_bytes = base64.b64decode(encoded)
# except Exception as e:
# return jsonify({
# 'success': False,
# 'message': f'视频数据解码失败: {str(e)}'
# }), 400
try:
header, encoded = video_data.split(',', 1)
video_bytes = base64.b64decode(encoded)
# with open(r'D:/111.webm', 'wb') as f:
# f.write(video_bytes)
except Exception as e:
return jsonify({
'success': False,
'message': f'视频数据解码失败: {str(e)}'
}), 400
# 停止同步录制，传递视频数据
try:
logger.debug(f'调用device_manager.stop_recording，session_id: {session_id}, video_data长度: {len(video_data) if video_data else 0}')
@ -716,7 +747,7 @@ def stop_detection(session_id):
logger.warning(f'视频数据为空，session_id: {session_id}')
else:
logger.debug(f'视频数据长度: {len(video_data)} 字符，约 {len(video_data)*3/4/1024:.2f} KB, session_id: {session_id}')
restrt=device_manager.stop_recording(session_id, video_data_base64=video_data)
restrt=device_manager.stop_recording(session_id, video_data_base64=video_bytes)
logger.error(restrt)
except Exception as rec_e:
logger.error(f'停止同步录制失败: {rec_e}', exc_info=True)
@ -786,7 +817,7 @@ def collect_detection_data(session_id):
# 获取请求数据
data = flask_request.get_json() or {}
patient_id = data.get('patient_id')
screen_image_base64 = data.get('screen_image')
screen_image_base64 = data.get('imageData')
# 如果没有提供patient_id，从会话信息中获取
if not patient_id:
@ -871,14 +902,29 @@ def stop_sync_recording():
data = flask_request.get_json()
session_id = data.get('session_id')
video_data = data.get('videoData') # 新增接收前端传递的视频数据
if not video_data.startswith('data:video/'):
return jsonify({
'success': False,
'message': '无效的视频数据格式'
}), 400
# 提取base64数据
try:
import base64
header, encoded = video_data.split(',', 1)
video_bytes = base64.b64decode(encoded)
except Exception as e:
return jsonify({
'success': False,
'message': f'视频数据解码失败: {str(e)}'
}), 400
if not session_id:
return jsonify({
'success': False,
'error': '缺少必要参数: session_id'
}), 400
result = device_manager.stop_recording(session_id, video_data_base64=video_data)
result = device_manager.stop_recording(session_id, video_data_base64=video_bytes)
if result['success']:
logger.info(f'同步录制已停止 - 会话ID: {session_id}')

400
backend/device_manager.py
View File

@ -31,7 +31,7 @@ from database import DatabaseManager
try:
import pykinect_azure as pykinect
# 重新启用FemtoBolt功能，使用正确的Orbbec SDK K4A Wrapper路径
FEMTOBOLT_AVAILABLE = False
FEMTOBOLT_AVAILABLE = True
print("信息: pykinect_azure库已安装，FemtoBolt深度相机功能已启用")
print("使用Orbbec SDK K4A Wrapper以确保与FemtoBolt设备的兼容性")
except ImportError:
@ -57,6 +57,7 @@ class DeviceManager:
}
self.calibration_data = {}
self.data_lock = threading.Lock()
self.camera_lock = threading.Lock() # 摄像头访问锁
self.latest_data = {}
# 数据库连接
@ -69,6 +70,12 @@ class DeviceManager:
self.femtobolt_streaming_thread = None
self.streaming_stop_event = threading.Event()
# 全局帧缓存机制
self.frame_cache = {}
self.frame_cache_lock = threading.RLock() # 可重入锁
self.max_cache_size = 10 # 最大缓存帧数
self.cache_timeout = 5.0 # 缓存超时时间（秒）
# 同步录制状态
self.sync_recording = False
self.current_session_id = None
@ -104,6 +111,8 @@ class DeviceManager:
# 初始化设备
self._init_devices()
def _init_devices(self):
"""初始化所有设备"""
@ -141,6 +150,8 @@ class DeviceManager:
self.camera.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
self.camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
self.camera.set(cv2.CAP_PROP_FPS, 30)
# 设置缓冲区大小为1，避免帧积累
self.camera.set(cv2.CAP_PROP_BUFFERSIZE, 1)
self.device_status['camera'] = True
logger.info(f'摄像头初始化成功，设备索引: {device_index}')
@ -334,9 +345,11 @@ class DeviceManager:
"""刷新设备连接"""
logger.info('刷新设备连接...')
# 重新初始化所有设备
if self.camera:
self.camera.release()
# 使用锁保护摄像头重新初始化
with self.camera_lock:
if self.camera:
self.camera.release()
self.camera = None
self._init_devices()
@ -484,6 +497,8 @@ class DeviceManager:
return {'status': 'failed', 'error': str(e)}
def collect_data(self, session_id: str, patient_id: str, screen_image_base64: str = None) -> Dict[str, Any]:
# 实例化VideoStreamManager（VideoStreamManager类在同一文件中定义）
video_stream_manager = VideoStreamManager()
"""采集所有设备数据并保存到指定目录结构
Args:
@ -529,12 +544,15 @@ class DeviceManager:
# data['body_pose'] = json.dumps(body_pose_data)
# logger.debug(f'身体姿态数据采集成功: {session_id}')
# # 3. 采集身体视频截图（从FemtoBolt深度相机获取）
# if self.device_status['femtobolt']:
# body_image_path = self._capture_body_image(data_dir)
# if body_image_path:
# data['body_image'] = str(body_image_path)
# logger.debug(f'身体截图保存成功: {body_image_path}')
# 3. 采集身体视频截图（从FemtoBolt深度相机获取）
if self.device_status['femtobolt']:
try:
body_image_path = video_stream_manager._capture_body_image(data_dir, self)
if body_image_path:
data['body_image'] = str(body_image_path)
logger.debug(f'身体截图保存成功: {body_image_path}')
except Exception as e:
logger.error(f'调用_video_stream_manager._capture_body_image异常: {e}')
# # 4. 采集足部压力数据（从压力传感器获取）
# if self.device_status['pressure']:
@ -543,12 +561,12 @@ class DeviceManager:
# data['foot_data'] = json.dumps(foot_data)
# logger.debug(f'足部压力数据采集成功: {session_id}')
# # 5. 采集足部监测视频截图（从摄像头获取）
# if self.device_status['camera']:
# foot_image_path = self._capture_foot_image(data_dir)
# if foot_image_path:
# data['foot_image'] = str(foot_image_path)
# logger.debug(f'足部截图保存成功: {foot_image_path}')
# 5. 采集足部监测视频截图（从摄像头获取）
if self.device_status['camera']:
foot_image_path = video_stream_manager._capture_foot_image(data_dir, self)
if foot_image_path:
data['foot_image'] = str(foot_image_path)
logger.debug(f'足部截图保存成功: {foot_image_path}')
# # 6. 生成足底压力数据图（从压力传感器数据生成）
# if self.device_status['pressure']:
@ -560,7 +578,7 @@ class DeviceManager:
# 7. 保存屏幕录制截图（从前端传入的base64数据）
if screen_image_base64:
try:
logger.debug(f'屏幕截图保存.................{screen_image_base64}')
# logger.debug(f'屏幕截图保存.................{screen_image_base64}')
# 保存屏幕截图的base64数据为图片文件
screen_image_path = None
if screen_image_base64:
@ -978,37 +996,88 @@ class DeviceManager:
def _camera_streaming_thread(self):
"""足部监视摄像头推流线程"""
frame_count = 0
consecutive_failures = 0
max_consecutive_failures = 10
try:
while self.camera_streaming and not self.streaming_stop_event.is_set():
if self.camera and self.camera.isOpened():
ret, frame = self.camera.read()
if ret and self.socketio:
# 编码并推送帧
try:
# 调整帧大小以减少网络负载
height, width = frame.shape[:2]
if width > 640:
scale = 640 / width
new_width = 640
new_height = int(height * scale)
frame = cv2.resize(frame, (new_width, new_height))
if self.camera:
# 使用摄像头锁避免与录制和截图功能冲突
with self.camera_lock:
# 检查摄像头状态
if not self.camera.isOpened():
logger.warning('推流线程检测到摄像头已关闭，尝试重新打开')
device_index = 0
if self.db_manager:
try:
monitor_config = self.db_manager.get_system_setting('monitor_device_index')
if monitor_config:
device_index = int(monitor_config)
except Exception:
pass
# JPEG编码
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 80]
success, buffer = cv2.imencode('.jpg', frame, encode_param)
if success:
jpg_as_text = base64.b64encode(buffer).decode('utf-8')
self.socketio.emit('video_frame', {
'image': jpg_as_text,
'frame_id': frame_count,
'timestamp': time.time()
})
frame_count += 1
self.camera.open(device_index)
if self.camera.isOpened():
# 重新设置摄像头参数
self.camera.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
self.camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
self.camera.set(cv2.CAP_PROP_FPS, 30)
self.camera.set(cv2.CAP_PROP_BUFFERSIZE, 1)
logger.info('推流线程摄像头重新打开成功')
consecutive_failures = 0
else:
logger.error('推流线程摄像头重新打开失败')
consecutive_failures += 1
time.sleep(0.5)
continue
except Exception as e:
logger.debug(f'摄像头帧推送失败: {e}')
ret, frame = self.camera.read()
if ret and frame is not None:
# 保存原始帧到全局缓存
self._save_frame_to_cache(frame, 'camera')
if self.socketio:
# 编码并推送帧
try:
# 调整帧大小以减少网络负载
display_frame = frame.copy()
height, width = display_frame.shape[:2]
if width > 640:
scale = 640 / width
new_width = 640
new_height = int(height * scale)
display_frame = cv2.resize(display_frame, (new_width, new_height))
# JPEG编码
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 80]
success, buffer = cv2.imencode('.jpg', display_frame, encode_param)
if success:
jpg_as_text = base64.b64encode(buffer).decode('utf-8')
self.socketio.emit('video_frame', {
'image': jpg_as_text,
'frame_id': frame_count,
'timestamp': time.time()
})
frame_count += 1
consecutive_failures = 0 # 重置失败计数
except Exception as e:
consecutive_failures += 1
if consecutive_failures <= 3:
logger.debug(f'摄像头帧推送失败 (连续失败{consecutive_failures}次): {e}')
else:
consecutive_failures += 1
if consecutive_failures <= 3:
logger.warning(f"推流线程无法从足部摄像头获取帧 (连续失败{consecutive_failures}次)")
elif consecutive_failures == max_consecutive_failures:
logger.error(f"推流线程足部摄像头连续失败{max_consecutive_failures}次，可能需要重启设备")
time.sleep(0.1) # 短暂等待
else:
time.sleep(0.1) # 摄像头不可用时等待
# 控制帧率
# time.sleep(1/30) # 30 FPS
@ -1388,7 +1457,7 @@ class DeviceManager:
return result
def stop_recording(self, session_id: str, video_data_base64: str = None) -> Dict[str, Any]:
def stop_recording(self, session_id: str, video_data_base64) -> Dict[str, Any]:
"""停止同步录制
Args:
@ -1424,7 +1493,7 @@ class DeviceManager:
# 定义视频文件路径
feet_video_path = os.path.join(base_path, 'feet.mp4')
body_video_path = os.path.join(base_path, 'body.mp4')
screen_video_path = os.path.join(base_path, 'screen.mp4')
screen_video_path = os.path.join(base_path, 'screen.webm')
# 等待录制线程结束
threads_to_join = [
@ -1445,19 +1514,24 @@ class DeviceManager:
# 清理视频写入器并收集文件信息
video_files = self._cleanup_video_writers()
# 保存传入的屏幕录制视频数据，替代原有屏幕录制视频保存逻辑
if video_data_base64:
try:
video_bytes = base64.b64decode(video_data_base64)
with open(screen_video_path, 'wb') as f:
f.write(video_bytes)
video_files.append(screen_video_path)
logger.info(f'屏幕录制视频保存成功，路径: {screen_video_path}, 文件大小: {os.path.getsize(screen_video_path)} 字节')
except Exception as e:
logger.error(f'保存屏幕录制视频失败: {e}', exc_info=True)
logger.debug(f'视频数据长度: {len(video_data_base64)}')
raise
# video_bytes = base64.b64decode(video_data_base64)
with open(screen_video_path, 'wb') as f:
f.write(video_data_base64)
video_files.append(screen_video_path)
logger.info(f'屏幕录制视频保存成功，路径: {screen_video_path}, 文件大小: {os.path.getsize(screen_video_path)} 字节')
# # 保存传入的屏幕录制视频数据，替代原有屏幕录制视频保存逻辑
# if video_data_base64:
# try:
# # video_bytes = base64.b64decode(video_data_base64)
# with open(screen_video_path, 'wb') as f:
# f.write(video_data_base64)
# video_files.append(screen_video_path)
# logger.info(f'屏幕录制视频保存成功，路径: {screen_video_path}, 文件大小: {os.path.getsize(screen_video_path)} 字节')
# except Exception as e:
# logger.error(f'保存屏幕录制视频失败: {e}', exc_info=True)
# logger.debug(f'视频数据长度: {len(video_data_base64)}')
# raise
result['video_files'] = video_files
@ -1507,12 +1581,35 @@ class DeviceManager:
def _feet_recording_thread(self):
"""足部视频录制线程"""
consecutive_failures = 0
max_consecutive_failures = 10
try:
while self.sync_recording and not self.recording_stop_event.is_set():
if self.camera and self.camera.isOpened() and self.feet_video_writer:
ret, frame = self.camera.read()
if ret:
if self.feet_video_writer:
# 从全局缓存获取最新帧
frame, frame_timestamp = self._get_latest_frame_from_cache('camera')
if frame is not None:
# 写入录制文件
self.feet_video_writer.write(frame)
consecutive_failures = 0 # 重置失败计数
# 记录录制统计
if hasattr(self, 'recording_frame_count'):
self.recording_frame_count += 1
else:
self.recording_frame_count = 1
else:
consecutive_failures += 1
if consecutive_failures <= 3:
logger.warning(f"录制线程无法从缓存获取帧 (连续失败{consecutive_failures}次)")
elif consecutive_failures == max_consecutive_failures:
logger.error(f"录制线程连续失败{max_consecutive_failures}次，可能缓存无数据或推流已停止")
# 等待一段时间再重试
time.sleep(0.1)
time.sleep(1/30) # 30 FPS
@ -1606,6 +1703,13 @@ class DeviceManager:
return video_files
def __del__(self):
"""析构函数，确保资源被正确释放"""
try:
self.cleanup()
except Exception as e:
logger.error(f'析构函数清理资源失败: {e}')
def cleanup(self):
"""清理资源"""
try:
@ -1616,8 +1720,14 @@ class DeviceManager:
if self.sync_recording:
self.stop_recording(self.current_session_id)
if self.camera:
self.camera.release()
# 使用锁保护摄像头释放
with self.camera_lock:
if self.camera:
self.camera.release()
self.camera = None
logger.debug('摄像头资源已释放')
if hasattr(self, 'video_writer') and self.video_writer:
self.video_writer.release()
@ -1640,6 +1750,14 @@ class DeviceManager:
if self.femtobolt_camera:
self.femtobolt_camera = None
# 清理帧缓存
try:
with self.frame_cache_lock:
self.frame_cache.clear()
logger.debug('帧缓存已清理')
except Exception as cache_error:
logger.error(f'清理帧缓存失败: {cache_error}')
logger.debug('设备资源已清理')
except Exception as e:
@ -2197,18 +2315,112 @@ class VideoStreamManager:
logger.error(f'身体姿态数据采集失败: {e}')
return None
def _capture_body_image(self, data_dir: Path) -> Optional[str]:
def _capture_body_image(self, data_dir: Path, device_manager) -> Optional[str]:
"""采集身体视频截图（从FemtoBolt深度相机获取）"""
try:
# 模拟从FemtoBolt深度相机获取图像
# 实际实现中应该从深度相机获取真实图像
image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
image = None
# 检查是否有device_manager实例且FemtoBolt深度相机可用
if (device_manager is not None and
FEMTOBOLT_AVAILABLE and
hasattr(device_manager, 'femtobolt_camera') and
device_manager.femtobolt_camera is not None):
# 从FemtoBolt深度相机获取真实图像
logger.info('正在从FemtoBolt深度相机获取身体图像...')
capture = device_manager.femtobolt_camera.update()
if capture is not None:
# 获取深度图像
ret, depth_image = capture.get_depth_image()
if ret and depth_image is not None:
# 读取config.ini中的深度范围配置
import configparser
config = configparser.ConfigParser()
config.read('config.ini')
try:
depth_range_min = int(config.get('DEFAULT', 'femtobolt_depth_range_min', fallback='1400'))
depth_range_max = int(config.get('DEFAULT', 'femtobolt_depth_range_max', fallback='1900'))
except Exception:
depth_range_min = None
depth_range_max = None
# 优化深度图彩色映射，范围外用黑色，区间内用Jet模型从蓝色到黄色到红色渐变
if depth_range_min is not None and depth_range_max is not None:
# 归一化深度值到0-255范围
depth_normalized = np.clip(depth_image, depth_range_min, depth_range_max)
depth_normalized = ((depth_normalized - depth_range_min) / (depth_range_max - depth_range_min) * 255).astype(np.uint8)
# 应用OpenCV的COLORMAP_JET进行伪彩色映射
depth_colored = cv2.applyColorMap(depth_normalized, cv2.COLORMAP_JET)
# 范围外用黑色
mask_outside = (depth_image < depth_range_min) | (depth_image > depth_range_max)
depth_colored[mask_outside] = [0, 0, 0] # BGR黑色
else:
# 如果没有配置，使用默认伪彩色映射
depth_colored = cv2.convertScaleAbs(depth_image, alpha=0.03)
depth_colored = cv2.applyColorMap(depth_colored, cv2.COLORMAP_JET)
# 转换颜色格式（如果需要）
if len(depth_colored.shape) == 3 and depth_colored.shape[2] == 4:
depth_colored = cv2.cvtColor(depth_colored, cv2.COLOR_BGRA2BGR)
elif len(depth_colored.shape) == 3 and depth_colored.shape[2] == 3:
pass
# 预处理：裁剪成宽460，高819，保持高度不裁剪，宽度从中间裁剪
height, width = depth_colored.shape[:2]
target_width = 460
target_height = 819
# 计算宽度裁剪起点
if width > target_width:
left = (width - target_width) // 2
right = left + target_width
cropped_image = depth_colored[:, left:right]
else:
cropped_image = depth_colored
# 如果高度不足target_height，进行上下填充黑边
cropped_height = cropped_image.shape[0]
if cropped_height < target_height:
pad_top = (target_height - cropped_height) // 2
pad_bottom = target_height - cropped_height - pad_top
cropped_image = cv2.copyMakeBorder(cropped_image, pad_top, pad_bottom, 0, 0, cv2.BORDER_CONSTANT, value=[0,0,0])
elif cropped_height > target_height:
# 如果高度超过target_height，裁剪高度中间部分
top = (cropped_height - target_height) // 2
cropped_image = cropped_image[top:top+target_height, :]
# 最终调整大小，保持宽460，高819
image = cv2.resize(cropped_image, (target_width, target_height))
logger.info(f'成功获取FemtoBolt深度图像，尺寸: {image.shape}')
else:
logger.warning('无法从FemtoBolt获取深度图像，使用模拟图像')
# 使用模拟图像作为备用
image = np.zeros((819, 460, 3), dtype=np.uint8)
cv2.rectangle(image, (50, 50), (410, 769), (0, 255, 0), 2)
cv2.putText(image, 'FemtoBolt Unavailable', (75, 400), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
else:
logger.warning('FemtoBolt capture为None，使用模拟图像')
# 使用模拟图像作为备用
image = np.zeros((819, 460, 3), dtype=np.uint8)
cv2.rectangle(image, (50, 50), (410, 769), (0, 255, 0), 2)
cv2.putText(image, 'Capture Failed', (120, 400), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
else:
logger.warning('FemtoBolt深度相机不可用，使用模拟图像')
# 使用模拟图像作为备用
image = np.zeros((819, 460, 3), dtype=np.uint8)
cv2.rectangle(image, (50, 50), (410, 769), (0, 255, 0), 2)
cv2.putText(image, 'Camera Not Available', (60, 400), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
# 保存图片
image_path = data_dir / 'body_image.jpg'
cv2.imwrite(str(image_path), image)
logger.info(f'身体图像已保存到: {image_path}')
return str(image_path.relative_to(Path.cwd()))
return image_path
except Exception as e:
logger.error(f'身体截图保存失败: {e}')
return None
@ -2251,26 +2463,54 @@ class VideoStreamManager:
logger.error(f'足部压力数据采集失败: {e}')
return None
def _capture_foot_image(self, data_dir: Path) -> Optional[str]:
"""采集足部监测视频截图（从摄像头获取）"""
def _capture_foot_image(self, data_dir: Path, device_manager) -> Optional[str]:
"""采集足部监测视频截图（从全局缓存获取）"""
try:
if self.camera is not None:
ret, frame = self.camera.read()
if ret:
# 保存图片
image_path = data_dir / 'foot_image.jpg'
cv2.imwrite(str(image_path), frame)
return str(image_path.relative_to(Path.cwd()))
image = None
# 如果摄像头不可用，生成模拟图像
image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
# 检查是否有device_manager实例
if device_manager is not None:
logger.info('正在从全局缓存获取最新图像...')
# 从全局缓存获取最新帧
frame, frame_timestamp = device_manager._get_latest_frame_from_cache('camera')
if frame is not None:
# 使用缓存中的图像
image = frame.copy() # 复制帧数据避免引用问题
current_time = time.time()
frame_age = current_time - frame_timestamp if frame_timestamp else 0
logger.info(f'成功获取缓存图像，尺寸: {image.shape}，帧龄: {frame_age:.2f}秒')
else:
logger.warning('缓存中无可用图像，使用模拟图像')
image = np.zeros((480, 640, 3), dtype=np.uint8)
cv2.rectangle(image, (50, 50), (590, 430), (0, 255, 0), 2)
cv2.putText(image, 'No Cached Frame', (120, 250), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
else:
logger.warning('设备管理器不可用，使用模拟图像')
# 使用模拟图像作为备用
image = np.zeros((480, 640, 3), dtype=np.uint8)
cv2.rectangle(image, (50, 50), (590, 430), (0, 255, 0), 2)
cv2.putText(image, 'Device Manager N/A', (100, 250), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
# 保存图片
image_path = data_dir / 'foot_image.jpg'
cv2.imwrite(str(image_path), image)
logger.info(f'足部图像已保存到: {image_path}')
return str(image_path.relative_to(Path.cwd()))
return image_path
except Exception as e:
logger.error(f'足部截图保存失败: {e}')
return None
# 即使出错也要保存一个模拟图像
try:
image = np.zeros((480, 640, 3), dtype=np.uint8)
cv2.rectangle(image, (50, 50), (590, 430), (255, 0, 0), 2)
cv2.putText(image, 'Error Occurred', (180, 250), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2)
image_path = data_dir / 'foot_image.jpg'
cv2.imwrite(str(image_path), image)
return image_path
except Exception:
return None
def _generate_foot_pressure_image(self, data_dir: Path) -> Optional[str]:
"""生成足底压力数据图（从压力传感器数据生成）"""

4
frontend/src/renderer/src/services/api.js
View File

@ -16,7 +16,7 @@ api.interceptors.request.use(
if (window.electronAPI) {
config.baseURL = window.electronAPI.getBackendUrl()
} else {
config.baseURL = 'http://192.168.1.58:5000'
config.baseURL = 'http://192.168.1.173:5000'
}
// 添加时间戳防止缓存
@ -599,7 +599,7 @@ export const getBackendUrl = () => {
if (window.electronAPI) {
return window.electronAPI.getBackendUrl()
} else {
return 'http://192.168.1.58:5000'
return 'http://192.168.1.173:5000'
}
}