#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AppServer类 - 主应用服务器
实现Flask和SocketIO服务，替代app.py功能
"""
import os
import sys
import json
import time
import threading
from datetime import datetime
from flask import Flask, jsonify
from flask import request as flask_request
from flask_cors import CORS
import logging
from flask_socketio import SocketIO, emit
import configparser
import argparse
# 添加当前目录到路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# 导入模块
from database import DatabaseManager
from utils import config as app_config
# 导入设备管理器
try:
from devices.camera_manager import CameraManager
from devices.imu_manager import IMUManager
from devices.pressure_manager import PressureManager
from devices.femtobolt_manager import FemtoBoltManager
from devices.device_coordinator import DeviceCoordinator
from devices.utils.config_manager import ConfigManager
except ImportError:
# 如果上面的导入失败，尝试直接导入
from camera_manager import CameraManager
import imu_manager
import pressure_manager
import femtobolt_manager
import device_coordinator
from utils import config_manager
IMUManager = imu_manager.IMUManager
PressureManager = pressure_manager.PressureManager
FemtoBoltManager = femtobolt_manager.FemtoBoltManager
DeviceCoordinator = device_coordinator.DeviceCoordinator
ConfigManager = config_manager.ConfigManager
class AppServer:
"""主应用服务器类"""
def __init__(self, host='localhost', port=5000, debug=False):
"""
初始化应用服务器
Args:
host: 服务器主机
port: 服务器端口
debug: 调试模式
"""
self.host = host
self.port = port
self.debug = debug
# 初始化日志
self._init_logging()
# Flask应用
self.app = Flask(__name__)
self.app.config['SECRET_KEY'] = 'body-balance-detection-system-2024'
# SocketIO
self._init_socketio()
# CORS配置
CORS(self.app, origins='*', supports_credentials=True,
allow_headers=['Content-Type', 'Authorization'],
methods=['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS'])
# 配置管理
self._init_config()
# 全局变量
self.db_manager = None
self.current_detection = None
self.detection_thread = None
# 数据推送状态
self.is_pushing_data = False
# 设备管理器
self.config_manager = None
self.device_coordinator = None
self.device_managers = {
'camera': None,
'femtobolt': None,
'imu': None,
'pressure': None
}
# 注册路由和事件
self._register_routes()
self._register_socketio_events()
def _init_logging(self):
"""初始化日志配置"""
# 日志目录
if getattr(sys, 'frozen', False):
# 打包后的可执行文件
log_dir = os.path.join(os.path.dirname(sys.executable), 'logs')
else:
# 开发环境
log_dir = 'logs'
# 创建日志目录
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, 'backend.log')
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(log_file, encoding='utf-8'),
logging.StreamHandler()
]
)
# 设置werkzeug日志级别为WARNING，过滤掉INFO级别的访问日志
logging.getLogger('werkzeug').setLevel(logging.WARNING)
self.logger = logging.getLogger(__name__)
def _init_socketio(self):
"""初始化SocketIO"""
try:
self.socketio = SocketIO(
self.app,
cors_allowed_origins='*',
async_mode='threading',
#async_mode='eventlet',
logger=False,
engineio_logger=False,
ping_timeout=60,
ping_interval=25,
manage_session=False,
always_connect=False,
transports=['polling', 'websocket'], # 优先使用polling
allow_upgrades=True, # 允许升级到websocket
cookie=None # 禁用cookie
)
self.logger.info('SocketIO初始化成功')
except Exception as e:
self.logger.error(f'SocketIO初始化失败: {e}')
self.socketio = None
# 设置SocketIO日志级别
logging.getLogger('socketio').setLevel(logging.WARNING)
logging.getLogger('engineio').setLevel(logging.WARNING)
def _init_config(self):
"""初始化配置"""
self.config = configparser.ConfigParser()
# 配置文件路径
if getattr(sys, 'frozen', False):
# 打包后的可执行文件
config_path = os.path.join(os.path.dirname(sys.executable), 'config.ini')
else:
# 开发环境
config_path = os.path.join(os.path.dirname(__file__), 'config.ini')
self.config.read(config_path, encoding='utf-8')
device_index = self.config.get('CAMERA', 'device_index', fallback=None)
print(f"设备号: {device_index}")
def init_app(self):
"""初始化应用组件"""
try:
# 初始化数据库管理器
self.logger.info('正在初始化数据库管理器...')
# 在打包环境中使用可执行文件的目录，在开发环境中使用脚本文件的目录
if getattr(sys, 'frozen', False):
# 打包环境
base_dir = os.path.dirname(sys.executable)
else:
# 开发环境
base_dir = os.path.dirname(__file__)
db_path = os.path.join(base_dir, 'data', 'body_balance.db')
os.makedirs(os.path.dirname(db_path), exist_ok=True)
self.db_manager = DatabaseManager(db_path)
self.db_manager.init_database()
self.logger.info('数据库管理器初始化完成')
# 初始化配置管理器
self.logger.info('正在初始化配置管理器...')
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('设备协调器初始化完成')
# 启动Flask应用
host = self.host
port = self.port
debug = self.debug
self.logger.info(f'启动Flask应用 - Host: {host}, Port: {port}, Debug: {debug}')
if self.socketio:
self.socketio.run(self.app, host=host, port=port, debug=debug, allow_unsafe_werkzeug=True)
else:
self.app.run(host=host, port=port, debug=debug)
except Exception as e:
self.logger.error(f'应用初始化失败: {e}')
raise
def _register_routes(self):
"""注册Flask路由"""
# ==================== 基础API ====================
@self.app.route('/health', methods=['GET'])
def health_check():
"""健康检查"""
return jsonify({
'status': 'healthy',
'timestamp': datetime.now().isoformat(),
'version': '1.0.0'
})
@self.app.route('/test-socketio')
def test_socketio():
"""测试SocketIO连接"""
return 'SocketIO Test Page
'
@self.app.route('/api/health', methods=['GET'])
def api_health_check():
"""API健康检查"""
return jsonify({
'success': True,
'message': '后端服务运行正常',
'timestamp': datetime.now().isoformat(),
'database': self.db_manager is not None,
'config_manager': self.config_manager is not None,
'device_coordinator': self.device_coordinator is not None,
'device_managers': {name: manager is not None for name, manager in self.device_managers.items()}
})
# ==================== 认证API ====================
@self.app.route('/api/auth/login', methods=['POST'])
def login():
"""用户登录"""
try:
# 检查Content-Type
if not flask_request.is_json:
return jsonify({'success': False, 'message': '请求Content-Type必须为application/json'}), 415
data = flask_request.get_json(force=True)
if not data:
return jsonify({'success': False, 'message': '请求数据为空'}), 400
username = data.get('username')
password = data.get('password')
if not username or not password:
return jsonify({'success': False, 'message': '用户名和密码不能为空'}), 400
# 验证用户
user = self.db_manager.authenticate_user(username, password)
if user:
# 检查用户是否已激活
if not user['is_active']:
return jsonify({
'success': False,
'message': '账户未激活，请联系管理员审核'
}), 403
# 构建用户数据
user_data = {
'id': user['id'],
'username': user['username'],
'name': user['name'],
'role': 'admin' if user['user_type'] == 'admin' else 'user',
'user_type': user['user_type'],
'avatar': ''
}
# 生成token（实际项目中应使用JWT等安全token）
token = f"token_{user['username']}_{int(time.time())}"
self.logger.info(f'用户 {username} 登录成功')
return jsonify({
'success': True,
'data': {
'token': token,
'user': user_data
},
'message': '登录成功'
})
else:
self.logger.warning(f'用户 {username} 登录失败：用户名或密码错误')
return jsonify({
'success': False,
'message': '用户名或密码错误'
}), 401
except Exception as e:
self.logger.error(f'登录失败: {e}')
return jsonify({'success': False, 'message': '登录失败'}), 500
@self.app.route('/api/auth/register', methods=['POST'])
def register():
"""用户注册"""
try:
# 检查Content-Type
if not flask_request.is_json:
return jsonify({'success': False, 'message': '请求Content-Type必须为application/json'}), 415
data = flask_request.get_json(force=True)
username = data.get('username')
password = data.get('password')
name = data.get('name') or data.get('email', '')
phone = data.get('phone')
if not username or not password:
return jsonify({
'success': False,
'message': '用户名和密码不能为空'
}), 400
if len(password) < 6:
return jsonify({
'success': False,
'message': '密码长度不能少于6位'
}), 400
# 构建用户数据字典
user_data = {
'username': username,
'password': password,
'name': name,
'phone': phone
}
# 使用数据库注册用户
result = self.db_manager.register_user(user_data)
if result['success']:
self.logger.info(f'用户 {username} 注册成功，等待管理员审核')
return jsonify({
'success': True,
'message': '注册成功，请等待管理员审核后登录'
})
else:
return jsonify({
'success': False,
'message': result['message']
}), 400
except Exception as e:
self.logger.error(f'注册失败: {e}')
return jsonify({'success': False, 'message': '注册失败'}), 500
@self.app.route('/api/auth/logout', methods=['POST'])
def logout():
"""用户登出"""
try:
# 这里可以添加token失效逻辑
return jsonify({'success': True, 'message': '登出成功'})
except Exception as e:
self.logger.error(f'登出失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/auth/verify', methods=['GET'])
def verify_token():
"""验证token"""
try:
# 从请求头获取token
auth_header = flask_request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({
'success': False,
'message': '未提供有效的认证信息',
'data': {'valid': False}
}), 401
token = auth_header.split(' ')[1]
# 这里应该验证JWT token，简化处理
if token.startswith('token_'):
return jsonify({
'success': True,
'message': 'Token有效',
'data': {'valid': True}
})
else:
return jsonify({
'success': False,
'message': 'Token无效',
'data': {'valid': False}
}), 401
except Exception as e:
self.logger.error(f'Token验证失败: {e}')
return jsonify({
'success': False,
'message': '验证失败',
'data': {'valid': False}
}), 500
@self.app.route('/api/auth/forgot-password', methods=['POST'])
def forgot_password():
"""忘记密码 - 根据用户名和手机号找回密码"""
try:
data = flask_request.get_json()
username = data.get('username')
phone = data.get('phone')
if not username:
return jsonify({
'success': False,
'error': '请输入用户名'
}), 400
if not phone:
return jsonify({
'success': False,
'error': '请输入手机号码'
}), 400
# 验证手机号格式
import re
phone_pattern = r'^1[3-9]\d{9}$'
if not re.match(phone_pattern, phone):
return jsonify({
'success': False,
'error': '手机号格式不正确'
}), 400
# 查询用户信息
conn = self.db_manager.get_connection()
cursor = conn.cursor()
cursor.execute('''
SELECT username, password, phone FROM users
WHERE username = ? AND phone = ?
''', (username, phone))
user = cursor.fetchone()
if user:
# 用户存在且手机号匹配，返回数据库中存储的实际密码
actual_password = user['password']
self.logger.info(f'用户 {username} 密码查询成功')
return jsonify({
'success': True,
'password': actual_password, # 返回数据库中存储的实际密码
'message': '密码找回成功'
})
else:
# 检查用户是否存在
cursor.execute('SELECT username FROM users WHERE username = ?', (username,))
user_exists = cursor.fetchone()
if not user_exists:
return jsonify({
'success': False,
'error': '用户不存在'
}), 400
else:
return jsonify({
'success': False,
'error': '手机号不匹配'
}), 400
except Exception as e:
self.logger.error(f'忘记密码处理失败: {e}')
return jsonify({'success': False, 'error': '处理失败'}), 500
# ==================== 用户管理API ====================
@self.app.route('/api/users', methods=['GET'])
def get_users():
"""获取用户列表"""
try:
page = int(flask_request.args.get('page', 1))
size = int(flask_request.args.get('size', 10))
users = self.db_manager.get_users(page, size)
total = self.db_manager.get_users_count()
return jsonify({
'success': True,
'data': {
'users': users,
'total': total,
'page': page,
'size': size
}
})
except Exception as e:
self.logger.error(f'获取用户列表失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/users//approve', methods=['POST'])
def approve_user(user_id):
"""审核用户"""
try:
data = flask_request.get_json()
status = data.get('status') # 'approved' 或 'rejected'
if status not in ['approved', 'rejected']:
return jsonify({'success': False, 'error': '无效的审核状态'}), 400
result = self.db_manager.update_user_status(user_id, status)
if result:
return jsonify({
'success': True,
'message': f'用户已{"通过" if status == "approved" else "拒绝"}审核'
})
else:
return jsonify({'success': False, 'error': '用户不存在'}), 404
except Exception as e:
self.logger.error(f'审核用户失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/users/', methods=['DELETE'])
def delete_user(user_id):
"""删除用户"""
try:
result = self.db_manager.delete_user(user_id)
if result:
return jsonify({'success': True, 'message': '用户已删除'})
else:
return jsonify({'success': False, 'error': '用户不存在'}), 404
except Exception as e:
self.logger.error(f'删除用户失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
# ==================== 患者管理API ====================
@self.app.route('/api/patients', methods=['GET', 'POST'])
def handle_patients():
"""患者管理"""
if flask_request.method == 'GET':
# 获取患者列表
try:
page = int(flask_request.args.get('page', 1))
size = int(flask_request.args.get('size', 10))
search = flask_request.args.get('search', '')
patients = self.db_manager.get_patients(page, size, search)
total = self.db_manager.get_patients_count(search)
return jsonify({
'success': True,
'data': {
'patients': patients,
'total': total,
'page': page,
'size': size
}
})
except Exception as e:
self.logger.error(f'获取患者列表失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
elif flask_request.method == 'POST':
# 创建患者
try:
# 检查Content-Type
if not flask_request.is_json:
return jsonify({'success': False, 'message': '请求Content-Type必须为application/json'}), 415
data = flask_request.get_json(force=True)
required_fields = ['name', 'gender', 'age']
for field in required_fields:
if not data.get(field):
return jsonify({'success': False, 'error': f'{field}不能为空'}), 400
patient_data = {
'name': data['name'],
'gender': data['gender'],
'age': data['age'],
'birth_date': data.get('birth_date'),
'nationality': data.get('nationality'),
'residence': data.get('residence'),
'height': data.get('height'),
'weight': data.get('weight'),
'shoe_size': data.get('shoe_size'),
'phone': data.get('phone'),
'email': data.get('email'),
'occupation': data.get('occupation'),
'workplace': data.get('workplace'),
'medical_history': data.get('medical_history', ''),
'notes': data.get('notes', '')
}
patient_id = self.db_manager.create_patient(patient_data)
if patient_id:
return jsonify({
'success': True,
'message': '患者创建成功',
'data': {'patient_id': patient_id}
})
else:
return jsonify({'success': False, 'error': '患者创建失败'}), 500
except Exception as e:
self.logger.error(f'创建患者失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/patients/', methods=['GET', 'PUT', 'DELETE'])
def handle_patient(patient_id):
"""单个患者操作"""
if flask_request.method == 'GET':
# 获取患者详情
try:
patient = self.db_manager.get_patient(patient_id)
if patient:
return jsonify({'success': True, 'data': patient})
else:
return jsonify({'success': False, 'error': '患者不存在'}), 404
except Exception as e:
self.logger.error(f'获取患者详情失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
elif flask_request.method == 'PUT':
# 更新患者信息
try:
data = flask_request.get_json()
patient_data = {
'name': data.get('name'),
'gender': data.get('gender'),
'age': data.get('age'),
'birth_date': data.get('birth_date'),
'nationality': data.get('nationality'),
'residence': data.get('residence'),
'height': data.get('height'),
'weight': data.get('weight'),
'shoe_size': data.get('shoe_size'),
'phone': data.get('phone'),
'email': data.get('email'),
'occupation': data.get('occupation'),
'workplace': data.get('workplace'),
'medical_history': data.get('medical_history'),
'notes': data.get('notes')
}
self.db_manager.update_patient(patient_id, patient_data)
result = True
if result:
return jsonify({'success': True, 'message': '患者信息更新成功'})
else:
return jsonify({'success': False, 'error': '患者不存在'}), 404
except Exception as e:
self.logger.error(f'更新患者信息失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
elif flask_request.method == 'DELETE':
# 删除患者
try:
result = self.db_manager.delete_patient(patient_id)
if result:
return jsonify({'success': True, 'message': '患者已删除'})
else:
return jsonify({'success': False, 'error': '患者不存在'}), 404
except Exception as e:
self.logger.error(f'删除患者失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
# ==================== 设备管理API ====================
@self.app.route('/api/devices/status', methods=['GET'])
def get_device_status():
"""获取设备状态"""
try:
if self.device_coordinator:
status = self.device_coordinator.get_device_status()
return jsonify({'success': True, 'data': status})
else:
return jsonify({'success': False, 'error': '设备协调器未初始化'}), 500
except Exception as e:
self.logger.error(f'获取设备状态失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/devices/refresh', methods=['POST'])
def refresh_devices():
"""刷新设备"""
try:
if self.device_coordinator:
result = self.device_coordinator.refresh_all_devices()
return jsonify({'success': True, 'data': result})
else:
return jsonify({'success': False, 'error': '设备协调器未初始化'}), 500
except Exception as e:
self.logger.error(f'刷新设备失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
# ==================== 设备配置API ====================
@self.app.route('/api/config/devices', methods=['GET'])
def get_all_device_configs():
"""获取所有设备配置"""
try:
if self.config_manager:
configs = self.config_manager.get_all_device_configs()
return jsonify({
'success': True,
'data': configs
})
else:
return jsonify({'success': False, 'error': '配置管理器未初始化'}), 500
except Exception as e:
self.logger.error(f"获取设备配置失败: {e}")
return jsonify({
'success': False,
'message': f'获取设备配置失败: {str(e)}'
}), 500
@self.app.route('/api/config/devices/all', methods=['POST'])
def set_all_device_configs():
"""批量设置所有设备配置"""
try:
if not self.config_manager:
return jsonify({'success': False, 'error': '配置管理器未初始化'}), 500
data = flask_request.get_json()
if not data:
return jsonify({
'success': False,
'message': '请求数据不能为空'
}), 400
# 验证数据格式
supported_devices = ['imu', 'pressure', 'camera', 'femtobolt']
for device_name in data.keys():
if device_name not in supported_devices:
return jsonify({
'success': False,
'message': f'不支持的设备类型: {device_name}，支持的设备类型: {", ".join(supported_devices)}'
}), 400
result = self.config_manager.set_all_device_configs(data)
# 如果配置设置成功，异步重启设备数据推送
if result['success']:
def restart_devices_async():
"""异步重启设备数据推送"""
try:
self.logger.info("设备配置更新成功，异步重启设备数据推送...")
# 先停止当前的数据推送
if self.is_pushing_data:
self.stop_device_push_data()
time.sleep(1) # 等待停止完成
# 重新启动设备数据推送
self.start_device_push_data()
self.logger.info("设备配置更新并重启数据推送完成")
# 通过SocketIO通知前端重启完成
self.socketio.emit('device_restart_complete', {
'status': 'success',
'message': '设备重启完成'
}, namespace='/devices')
except Exception as restart_error:
self.logger.error(f"重启设备数据推送失败: {restart_error}")
# 通过SocketIO通知前端重启失败
self.socketio.emit('device_restart_complete', {
'status': 'error',
'message': f'设备重启失败: {str(restart_error)}'
}, namespace='/devices')
# 启动异步线程执行重启操作
restart_thread = threading.Thread(target=restart_devices_async)
restart_thread.daemon = True
restart_thread.start()
result['message'] = result.get('message', '') + ' 设备正在后台重启中，请稍候...'
status_code = 200 if result['success'] else 400
return jsonify(result), status_code
except Exception as e:
self.logger.error(f"批量设置设备配置失败: {e}")
return jsonify({
'success': False,
'message': f'批量设置设备配置失败: {str(e)}'
}), 500
@self.app.route('/api/devices/calibrate', methods=['POST'])
def calibrate_device():
"""校准设备"""
try:
data = flask_request.get_json()
device_type = data.get('device_type')
if not device_type:
return jsonify({'success': False, 'error': '设备类型不能为空'}), 400
if device_type in self.device_managers and self.device_managers[device_type]:
device_manager = self.device_managers[device_type]
if hasattr(device_manager, 'calibrate'):
result = device_manager.calibrate()
return jsonify({'success': True, 'data': result})
else:
return jsonify({'success': False, 'error': f'{device_type}设备不支持校准'}), 400
else:
return jsonify({'success': False, 'error': f'{device_type}设备管理器未初始化'}), 500
except Exception as e:
self.logger.error(f'校准设备失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/devices/imu/calibrate', methods=['POST'])
def calibrate_imu():
"""校准IMU"""
try:
if self.device_managers['imu']:
result = self.device_managers['imu'].calibrate()
if result:
return jsonify({'success': True, 'message': 'IMU校准成功'})
else:
return jsonify({'success': False, 'error': 'IMU校准失败'}), 500
else:
return jsonify({'success': False, 'error': 'IMU设备管理器未初始化'}), 500
except Exception as e:
self.logger.error(f'IMU校准失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
# ==================== 检测API ====================
@self.app.route('/api/detection/start', methods=['POST'])
def start_detection():
"""开始检测"""
try:
data = flask_request.get_json()
patient_id = data.get('patient_id')
detection_type = data.get('detection_type', 'balance')
duration = data.get('duration', 30)
if not patient_id:
return jsonify({'success': False, 'error': '患者ID不能为空'}), 400
# 检查是否已有检测在进行
if self.current_detection:
return jsonify({'success': False, 'error': '已有检测在进行中'}), 400
# 启动检测
detection_id = self.db_manager.create_detection_session(
patient_id=patient_id,
detection_type=detection_type,
duration=duration
)
if detection_id:
self.current_detection = {
'id': detection_id,
'patient_id': patient_id,
'type': detection_type,
'duration': duration,
'start_time': time.time()
}
# 启动检测线程
self.detection_thread = threading.Thread(
target=self._detection_worker,
args=(detection_id, duration)
)
self.detection_thread.start()
return jsonify({
'success': True,
'message': '检测已开始',
'data': {'detection_id': detection_id}
})
else:
return jsonify({'success': False, 'error': '创建检测会话失败'}), 500
except Exception as e:
self.logger.error(f'开始检测失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/detection/stop', methods=['POST'])
def stop_detection():
"""停止检测"""
try:
if not self.current_detection:
return jsonify({'success': False, 'error': '没有正在进行的检测'}), 400
# 停止检测
detection_id = self.current_detection['id']
self.db_manager.end_detection_session(detection_id)
self.current_detection = None
return jsonify({'success': True, 'message': '检测已停止'})
except Exception as e:
self.logger.error(f'停止检测失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/detection/status', methods=['GET'])
def get_detection_status():
"""获取检测状态"""
try:
if self.current_detection:
elapsed_time = time.time() - self.current_detection['start_time']
remaining_time = max(0, self.current_detection['duration'] - elapsed_time)
return jsonify({
'success': True,
'data': {
'is_detecting': True,
'detection_id': self.current_detection['id'],
'patient_id': self.current_detection['patient_id'],
'type': self.current_detection['type'],
'elapsed_time': elapsed_time,
'remaining_time': remaining_time,
'progress': min(100, (elapsed_time / self.current_detection['duration']) * 100)
}
})
else:
return jsonify({
'success': True,
'data': {'is_detecting': False}
})
except Exception as e:
self.logger.error(f'获取检测状态失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/detection/save-session', methods=['POST'])
def save_session_info():
"""保存会话信息"""
try:
data = flask_request.get_json()
session_id = data.get('session_id')
session_info = data.get('session_info', {})
if not session_id:
return jsonify({'success': False, 'error': '会话ID不能为空'}), 400
result = self.db_manager.save_session_info(session_id, session_info)
if result:
return jsonify({'success': True, 'message': '会话信息保存成功'})
else:
return jsonify({'success': False, 'error': '会话信息保存失败'}), 500
except Exception as e:
self.logger.error(f'保存会话信息失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/detection/collect-data', methods=['POST'])
def collect_detection_data():
"""采集检测数据"""
try:
data = flask_request.get_json()
session_id = data.get('session_id')
data_type = data.get('data_type')
data_content = data.get('data')
if not all([session_id, data_type, data_content]):
return jsonify({'success': False, 'error': '参数不完整'}), 400
result = self.db_manager.save_detection_data(session_id, data_type, data_content)
if result:
return jsonify({'success': True, 'message': '数据采集成功'})
else:
return jsonify({'success': False, 'error': '数据采集失败'}), 500
except Exception as e:
self.logger.error(f'采集检测数据失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
# ==================== 同步录制API ====================
@self.app.route('/api/sync-recording/start', methods=['POST'])
def start_sync_recording():
"""启动同步录制"""
try:
data = flask_request.get_json()
session_id = data.get('session_id')
if not session_id:
return jsonify({'success': False, 'error': '会话ID不能为空'}), 400
if self.device_manager:
result = self.device_manager.start_sync_recording(session_id)
if result['success']:
self.logger.info(f'同步录制已启动 - 会话ID: {session_id}')
return jsonify(result)
else:
return jsonify(result), 500
else:
return jsonify({'success': False, 'error': '设备管理器未初始化'}), 500
except Exception as e:
self.logger.error(f'启动同步录制失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
@self.app.route('/api/sync-recording/stop', methods=['POST'])
def stop_sync_recording():
"""停止同步录制"""
try:
data = flask_request.get_json()
session_id = data.get('session_id')
if not session_id:
return jsonify({'success': False, 'error': '会话ID不能为空'}), 400
if self.device_manager:
result = self.device_manager.stop_sync_recording(session_id)
if result['success']:
self.logger.info(f'同步录制已停止 - 会话ID: {session_id}')
return jsonify(result)
else:
return jsonify(result), 500
else:
return jsonify({'success': False, 'error': '设备管理器未初始化'}), 500
except Exception as e:
self.logger.error(f'停止同步录制失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
# ==================== 历史记录API ====================
@self.app.route('/api/history/sessions', methods=['GET'])
def get_detection_sessions():
"""获取检测会话历史"""
try:
page = int(flask_request.args.get('page', 1))
size = int(flask_request.args.get('size', 10))
patient_id = flask_request.args.get('patient_id')
sessions = self.db_manager.get_detection_sessions(page, size, patient_id)
total = self.db_manager.get_sessions_count(patient_id)
return jsonify({
'success': True,
'data': {
'sessions': sessions,
'total': total,
'page': page,
'size': size
}
})
except Exception as e:
self.logger.error(f'获取检测历史失败: {e}')
return jsonify({'success': False, 'error': str(e)}), 500
# ==================== 错误处理 ====================
@self.app.errorhandler(404)
def not_found(error):
return jsonify({'success': False, 'error': 'API接口不存在'}), 404
@self.app.errorhandler(500)
def internal_error(error):
return jsonify({'success': False, 'error': '服务器内部错误'}), 500
def _register_socketio_events(self):
"""注册SocketIO事件"""
if self.socketio is None:
return
# 注册统一设备命名空间的连接事件
@self.socketio.on('connect', namespace='/devices')
def handle_devices_connect():
self.logger.info('设备命名空间客户端连接')
emit('status', {'message': '设备命名空间连接成功'}, namespace='/devices')
# 连接时发送当前所有设备的状态
self.broadcast_all_device_status()
@self.socketio.on('disconnect', namespace='/devices')
def handle_devices_disconnect():
self.logger.info('设备命名空间客户端断开连接')
# 注册设备订阅事件
@self.socketio.on('subscribe_device', namespace='/devices')
def handle_subscribe_device(data):
"""订阅特定设备数据"""
device_type = data.get('device_type')
if device_type in ['camera', 'femtobolt', 'imu', 'pressure']:
self.logger.info(f'客户端订阅{device_type}设备数据')
emit('subscription_status', {
'device_type': device_type,
'status': 'subscribed',
'message': f'{device_type}设备数据订阅成功'
}, namespace='/devices')
else:
emit('subscription_status', {
'device_type': device_type,
'status': 'error',
'message': '不支持的设备类型'
}, namespace='/devices')
@self.socketio.on('unsubscribe_device', namespace='/devices')
def handle_unsubscribe_device(data):
"""取消订阅特定设备数据"""
device_type = data.get('device_type')
self.logger.info(f'客户端取消订阅{device_type}设备数据')
emit('subscription_status', {
'device_type': device_type,
'status': 'unsubscribed',
'message': f'{device_type}设备数据取消订阅成功'
}, namespace='/devices')
@self.socketio.on('start_push_data', namespace='/devices')
def handle_start_push_data():
"""启动数据推送"""
try:
self.start_device_push_data()
emit('test_status', {'status': 'started', 'message': '数据推送已开始'}, namespace='/devices')
except Exception as e:
emit('test_status', {'status': 'error', 'message': str(e)}, namespace='/devices')
@self.socketio.on('stop_push_data', namespace='/devices')
def handle_stop_push_data():
"""停止数据推送"""
try:
self.stop_device_push_data()
emit('test_status', {'status': 'stopped', 'message': '数据推送已停止'}, namespace='/devices')
except Exception as e:
emit('test_status', {'status': 'error', 'message': str(e)}, namespace='/devices')
def start_device_push_data(self):
"""开始设备数据推送"""
if self.is_pushing_data:
self.logger.warning('设备数据推送已在运行')
return
try:
self.logger.info('开始设备数据推送...')
self.is_pushing_data = True
# 并行启动真实设备管理器
failed_devices = []
device_threads = {}
device_results = {}
def initialize_device(device_name, manager):
"""设备初始化工作函数"""
try:
print(f"[DEBUG] 尝试初始化设备: {device_name}")
if manager.initialize():
print(f"[DEBUG] {device_name} 初始化成功，开始启动流")
manager.start_streaming()
device_results[device_name] = True
self.logger.info(f'{device_name}设备启动成功')
else:
print(f"[DEBUG] {device_name} 初始化失败")
device_results[device_name] = False
self.logger.error(f'{device_name}设备启动失败')
except Exception as e:
print(f"[DEBUG] {device_name} 初始化异常: {e}")
device_results[device_name] = False
self.logger.error(f'{device_name}设备启动异常: {e}')
# 为每个设备创建初始化线程
for device_name, manager in self.device_managers.items():
if manager is not None: # 确保管理器已初始化
thread = threading.Thread(
target=initialize_device,
args=(device_name, manager),
name=f'Init-{device_name}',
daemon=True
)
device_threads[device_name] = thread
thread.start()
else:
self.logger.warning(f'{device_name}管理器未初始化，跳过启动')
device_results[device_name] = False
# 等待所有设备初始化完成（最多等待30秒）
for device_name, thread in device_threads.items():
thread.join(timeout=30.0)
if thread.is_alive():
self.logger.warning(f'{device_name}设备初始化超时')
device_results[device_name] = False
# 收集失败的设备
for device_name, success in device_results.items():
if not success:
failed_devices.append(device_name)
# 输出启动结果摘要
successful_devices = [name for name, success in device_results.items() if success]
# 广播设备状态更新
for device_name, success in device_results.items():
self.broadcast_device_status(device_name, success)
if successful_devices:
self.logger.info(f'成功启动的设备: {", ".join(successful_devices)}')
if failed_devices:
self.logger.warning(f'启动失败的设备: {", ".join(failed_devices)}')
self.logger.info('设备数据推送已启动')
except Exception as e:
self.logger.error(f'启动设备数据推送失败: {e}')
self.is_pushing_data = False
raise
def stop_device_push_data(self):
"""停止设备数据推送"""
if not self.is_pushing_data:
self.logger.warning('设备数据推送未运行')
return
try:
self.logger.info('停止设备数据推送...')
self.is_pushing_data = False
# 停止设备管理器
for device_name, manager in self.device_managers.items():
if manager is not None:
try:
manager.stop_streaming()
manager.disconnect()
self.logger.info(f'{device_name}设备已停止')
# 广播设备状态为未连接
self.broadcast_device_status(device_name, False)
except Exception as e:
self.logger.error(f'停止{device_name}设备失败: {e}')
# 即使停止失败也广播为未连接状态
self.broadcast_device_status(device_name, False)
self.logger.info('设备数据推送已停止')
except Exception as e:
self.logger.error(f'停止设备数据推送失败: {e}')
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 broadcast_all_device_status(self):
"""广播所有设备状态"""
for device_name, manager in self.device_managers.items():
if manager is not None:
try:
# 检查设备是否连接（使用is_connected属性）
is_connected = hasattr(manager, 'is_connected') and getattr(manager, 'is_connected', False)
self.broadcast_device_status(device_name, is_connected)
except Exception as e:
self.logger.error(f'检查{device_name}设备状态失败: {e}')
self.broadcast_device_status(device_name, False)
def _on_device_status_change(self, device_name: str, is_connected: bool):
"""
设备状态变化回调函数
Args:
device_name: 设备名称
is_connected: 连接状态
"""
self.logger.info(f'设备状态变化: {device_name} -> {"已连接" if is_connected else "未连接"}')
self.broadcast_device_status(device_name, is_connected)
def _detection_worker(self, detection_id, duration):
"""检测工作线程"""
try:
self.logger.info(f'检测线程启动 - ID: {detection_id}, 持续时间: {duration}秒')
# 模拟检测过程
start_time = time.time()
while time.time() - start_time < duration:
if not self.current_detection:
break
# 发送检测进度
elapsed = time.time() - start_time
progress = min(100, (elapsed / duration) * 100)
if self.socketio:
self.socketio.emit('detection_progress', {
'detection_id': detection_id,
'progress': progress,
'elapsed_time': elapsed,
'remaining_time': max(0, duration - elapsed)
})
time.sleep(1)
# 检测完成
if self.current_detection and self.current_detection['id'] == detection_id:
self.db_manager.end_detection_session(detection_id)
self.current_detection = None
if self.socketio:
self.socketio.emit('detection_complete', {
'detection_id': detection_id,
'message': '检测已完成'
})
self.logger.info(f'检测完成 - ID: {detection_id}')
except Exception as e:
self.logger.error(f'检测线程异常: {e}')
if self.current_detection:
self.current_detection = None
def run(self, debug=None):
"""运行服务器"""
if debug is not None:
self.debug = debug
try:
self.logger.info('正在启动AppServer...')
self.init_app()
except KeyboardInterrupt:
self.logger.info('服务被用户中断')
except Exception as e:
self.logger.error(f'服务启动失败: {e}')
sys.exit(1)
finally:
self.logger.info('AppServer已停止')
def main():
"""主函数"""
# 解析命令行参数
parser = argparse.ArgumentParser(description='Body Balance Evaluation System Backend')
parser.add_argument('--host', default='localhost', help='Host address to bind to')
parser.add_argument('--port', type=int, default=5000, help='Port number to bind to')
parser.add_argument('--debug', action='store_true', help='Enable debug mode')
args = parser.parse_args()
# 创建并运行服务器
server = AppServer(host=args.host, port=args.port, debug=args.debug)
server.run()
if __name__ == '__main__':
main()