730 lines
31 KiB
Python
730 lines
31 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
数据处理模块
|
|
负责数据分析、报告生成和数据导出
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import numpy as np
|
|
import pandas as pd
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional, Any, Tuple
|
|
import logging
|
|
from pathlib import Path
|
|
import matplotlib.pyplot as plt
|
|
import matplotlib.dates as mdates
|
|
from matplotlib.backends.backend_pdf import PdfPages
|
|
import seaborn as sns
|
|
from reportlab.lib.pagesizes import letter, A4
|
|
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Image, Table, TableStyle
|
|
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
|
|
from reportlab.lib.units import inch
|
|
from reportlab.lib import colors
|
|
from reportlab.pdfgen import canvas
|
|
from reportlab.lib.utils import ImageReader
|
|
from io import BytesIO
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 设置中文字体
|
|
plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei']
|
|
plt.rcParams['axes.unicode_minus'] = False
|
|
sns.set_style("whitegrid")
|
|
|
|
class DataProcessor:
|
|
"""数据处理器"""
|
|
|
|
def __init__(self):
|
|
self.export_dir = Path('exports')
|
|
self.charts_dir = Path('charts')
|
|
|
|
# 创建必要的目录
|
|
self.export_dir.mkdir(exist_ok=True)
|
|
self.charts_dir.mkdir(exist_ok=True)
|
|
|
|
logger.info('数据处理器初始化完成')
|
|
|
|
def analyze_session(self, session_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""分析会话数据"""
|
|
try:
|
|
if not session_data or 'data' not in session_data:
|
|
return {'error': '没有可分析的数据'}
|
|
|
|
data_points = session_data['data']
|
|
if not data_points:
|
|
return {'error': '数据为空'}
|
|
|
|
# 数据预处理
|
|
processed_data = self._preprocess_session_data(data_points)
|
|
|
|
# 统计分析
|
|
statistical_analysis = self._statistical_analysis(processed_data)
|
|
|
|
# 趋势分析
|
|
trend_analysis = self._trend_analysis(processed_data)
|
|
|
|
# 异常检测
|
|
anomaly_analysis = self._anomaly_detection(processed_data)
|
|
|
|
# 生成图表
|
|
charts = self._generate_charts(processed_data, session_data['id'])
|
|
|
|
analysis_result = {
|
|
'session_info': {
|
|
'session_id': session_data['id'],
|
|
'patient_name': session_data.get('patient_name', '未知'),
|
|
'start_time': session_data.get('start_time'),
|
|
'end_time': session_data.get('end_time'),
|
|
'duration': session_data.get('duration'),
|
|
'data_points': len(data_points)
|
|
},
|
|
'statistical': statistical_analysis,
|
|
'trends': trend_analysis,
|
|
'anomalies': anomaly_analysis,
|
|
'charts': charts,
|
|
'summary': self._generate_summary(statistical_analysis, trend_analysis, anomaly_analysis)
|
|
}
|
|
|
|
return analysis_result
|
|
|
|
except Exception as e:
|
|
logger.error(f'会话数据分析失败: {e}')
|
|
return {'error': str(e)}
|
|
|
|
def _preprocess_session_data(self, data_points: List[Dict]) -> Dict[str, List]:
|
|
"""预处理会话数据"""
|
|
processed = {
|
|
'timestamps': [],
|
|
'pressure_left': [],
|
|
'pressure_right': [],
|
|
'pressure_total': [],
|
|
'balance_index': [],
|
|
'center_of_pressure_x': [],
|
|
'center_of_pressure_y': [],
|
|
'imu_pitch': [],
|
|
'imu_roll': [],
|
|
'imu_accel_total': []
|
|
}
|
|
|
|
for point in data_points:
|
|
try:
|
|
# 解析时间戳
|
|
timestamp = datetime.fromisoformat(point['timestamp'].replace('Z', '+00:00'))
|
|
processed['timestamps'].append(timestamp)
|
|
|
|
# 解析数据值
|
|
data_value = point['data_value']
|
|
|
|
if point['data_type'] == 'pressure':
|
|
processed['pressure_left'].append(data_value.get('left_foot', 0))
|
|
processed['pressure_right'].append(data_value.get('right_foot', 0))
|
|
processed['pressure_total'].append(data_value.get('total_pressure', 0))
|
|
processed['balance_index'].append(data_value.get('balance_index', 0))
|
|
|
|
cop = data_value.get('center_of_pressure', {'x': 0, 'y': 0})
|
|
processed['center_of_pressure_x'].append(cop.get('x', 0))
|
|
processed['center_of_pressure_y'].append(cop.get('y', 0))
|
|
|
|
elif point['data_type'] == 'imu':
|
|
processed['imu_pitch'].append(data_value.get('pitch', 0))
|
|
processed['imu_roll'].append(data_value.get('roll', 0))
|
|
processed['imu_accel_total'].append(data_value.get('total_accel', 0))
|
|
|
|
except Exception as e:
|
|
logger.warning(f'数据点处理失败: {e}')
|
|
continue
|
|
|
|
return processed
|
|
|
|
def _statistical_analysis(self, data: Dict[str, List]) -> Dict[str, Any]:
|
|
"""统计分析"""
|
|
try:
|
|
stats = {}
|
|
|
|
# 压力数据统计
|
|
if data['pressure_total']:
|
|
pressure_total = np.array(data['pressure_total'])
|
|
stats['pressure'] = {
|
|
'mean': float(np.mean(pressure_total)),
|
|
'std': float(np.std(pressure_total)),
|
|
'min': float(np.min(pressure_total)),
|
|
'max': float(np.max(pressure_total)),
|
|
'median': float(np.median(pressure_total))
|
|
}
|
|
|
|
# 左右脚压力分析
|
|
if data['pressure_left'] and data['pressure_right']:
|
|
left_pressure = np.array(data['pressure_left'])
|
|
right_pressure = np.array(data['pressure_right'])
|
|
|
|
stats['pressure_distribution'] = {
|
|
'left_mean': float(np.mean(left_pressure)),
|
|
'right_mean': float(np.mean(right_pressure)),
|
|
'left_ratio': float(np.mean(left_pressure) / np.mean(pressure_total)) if np.mean(pressure_total) > 0 else 0,
|
|
'right_ratio': float(np.mean(right_pressure) / np.mean(pressure_total)) if np.mean(pressure_total) > 0 else 0,
|
|
'asymmetry': float(abs(np.mean(left_pressure) - np.mean(right_pressure)) / np.mean(pressure_total)) if np.mean(pressure_total) > 0 else 0
|
|
}
|
|
|
|
# 平衡指数统计
|
|
if data['balance_index']:
|
|
balance_index = np.array(data['balance_index'])
|
|
stats['balance'] = {
|
|
'mean': float(np.mean(balance_index)),
|
|
'std': float(np.std(balance_index)),
|
|
'min': float(np.min(balance_index)),
|
|
'max': float(np.max(balance_index)),
|
|
'stability_score': float(1.0 - np.std(balance_index)) # 稳定性评分
|
|
}
|
|
|
|
# 重心位置统计
|
|
if data['center_of_pressure_x'] and data['center_of_pressure_y']:
|
|
cop_x = np.array(data['center_of_pressure_x'])
|
|
cop_y = np.array(data['center_of_pressure_y'])
|
|
|
|
stats['center_of_pressure'] = {
|
|
'mean_x': float(np.mean(cop_x)),
|
|
'mean_y': float(np.mean(cop_y)),
|
|
'std_x': float(np.std(cop_x)),
|
|
'std_y': float(np.std(cop_y)),
|
|
'range_x': float(np.max(cop_x) - np.min(cop_x)),
|
|
'range_y': float(np.max(cop_y) - np.min(cop_y)),
|
|
'total_displacement': float(np.sqrt(np.std(cop_x)**2 + np.std(cop_y)**2))
|
|
}
|
|
|
|
# IMU数据统计
|
|
if data['imu_pitch'] and data['imu_roll']:
|
|
pitch = np.array(data['imu_pitch'])
|
|
roll = np.array(data['imu_roll'])
|
|
|
|
stats['posture'] = {
|
|
'mean_pitch': float(np.mean(pitch)),
|
|
'mean_roll': float(np.mean(roll)),
|
|
'std_pitch': float(np.std(pitch)),
|
|
'std_roll': float(np.std(roll)),
|
|
'max_pitch': float(np.max(np.abs(pitch))),
|
|
'max_roll': float(np.max(np.abs(roll))),
|
|
'posture_stability': float(1.0 - (np.std(pitch) + np.std(roll)) / 20) # 姿态稳定性
|
|
}
|
|
|
|
return stats
|
|
|
|
except Exception as e:
|
|
logger.error(f'统计分析失败: {e}')
|
|
return {}
|
|
|
|
def _trend_analysis(self, data: Dict[str, List]) -> Dict[str, Any]:
|
|
"""趋势分析"""
|
|
try:
|
|
trends = {}
|
|
|
|
# 平衡指数趋势
|
|
if data['balance_index'] and len(data['balance_index']) > 10:
|
|
balance_trend = self._calculate_trend(data['balance_index'])
|
|
trends['balance_trend'] = {
|
|
'slope': balance_trend['slope'],
|
|
'direction': 'improving' if balance_trend['slope'] < 0 else 'deteriorating' if balance_trend['slope'] > 0 else 'stable',
|
|
'correlation': balance_trend['correlation']
|
|
}
|
|
|
|
# 压力分布趋势
|
|
if data['pressure_left'] and data['pressure_right'] and len(data['pressure_left']) > 10:
|
|
left_trend = self._calculate_trend(data['pressure_left'])
|
|
right_trend = self._calculate_trend(data['pressure_right'])
|
|
|
|
trends['pressure_trend'] = {
|
|
'left_slope': left_trend['slope'],
|
|
'right_slope': right_trend['slope'],
|
|
'asymmetry_trend': 'increasing' if abs(left_trend['slope'] - right_trend['slope']) > 0.1 else 'stable'
|
|
}
|
|
|
|
# 姿态趋势
|
|
if data['imu_pitch'] and data['imu_roll'] and len(data['imu_pitch']) > 10:
|
|
pitch_trend = self._calculate_trend(data['imu_pitch'])
|
|
roll_trend = self._calculate_trend(data['imu_roll'])
|
|
|
|
trends['posture_trend'] = {
|
|
'pitch_slope': pitch_trend['slope'],
|
|
'roll_slope': roll_trend['slope'],
|
|
'stability_trend': 'improving' if abs(pitch_trend['slope']) + abs(roll_trend['slope']) < 0.1 else 'deteriorating'
|
|
}
|
|
|
|
return trends
|
|
|
|
except Exception as e:
|
|
logger.error(f'趋势分析失败: {e}')
|
|
return {}
|
|
|
|
def _calculate_trend(self, values: List[float]) -> Dict[str, float]:
|
|
"""计算趋势"""
|
|
try:
|
|
x = np.arange(len(values))
|
|
y = np.array(values)
|
|
|
|
# 线性回归
|
|
slope, intercept = np.polyfit(x, y, 1)
|
|
correlation = np.corrcoef(x, y)[0, 1]
|
|
|
|
return {
|
|
'slope': float(slope),
|
|
'intercept': float(intercept),
|
|
'correlation': float(correlation)
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f'趋势计算失败: {e}')
|
|
return {'slope': 0, 'intercept': 0, 'correlation': 0}
|
|
|
|
def _anomaly_detection(self, data: Dict[str, List]) -> Dict[str, Any]:
|
|
"""异常检测"""
|
|
try:
|
|
anomalies = {}
|
|
|
|
# 检测平衡指数异常
|
|
if data['balance_index']:
|
|
balance_anomalies = self._detect_outliers(data['balance_index'])
|
|
anomalies['balance_anomalies'] = {
|
|
'count': len(balance_anomalies),
|
|
'percentage': len(balance_anomalies) / len(data['balance_index']) * 100,
|
|
'indices': balance_anomalies
|
|
}
|
|
|
|
# 检测压力异常
|
|
if data['pressure_total']:
|
|
pressure_anomalies = self._detect_outliers(data['pressure_total'])
|
|
anomalies['pressure_anomalies'] = {
|
|
'count': len(pressure_anomalies),
|
|
'percentage': len(pressure_anomalies) / len(data['pressure_total']) * 100,
|
|
'indices': pressure_anomalies
|
|
}
|
|
|
|
# 检测姿态异常
|
|
if data['imu_pitch'] and data['imu_roll']:
|
|
pitch_anomalies = self._detect_outliers(data['imu_pitch'])
|
|
roll_anomalies = self._detect_outliers(data['imu_roll'])
|
|
|
|
anomalies['posture_anomalies'] = {
|
|
'pitch_count': len(pitch_anomalies),
|
|
'roll_count': len(roll_anomalies),
|
|
'total_percentage': (len(pitch_anomalies) + len(roll_anomalies)) / (len(data['imu_pitch']) + len(data['imu_roll'])) * 100
|
|
}
|
|
|
|
return anomalies
|
|
|
|
except Exception as e:
|
|
logger.error(f'异常检测失败: {e}')
|
|
return {}
|
|
|
|
def _detect_outliers(self, values: List[float], threshold: float = 2.0) -> List[int]:
|
|
"""检测异常值"""
|
|
try:
|
|
data = np.array(values)
|
|
mean = np.mean(data)
|
|
std = np.std(data)
|
|
|
|
# 使用Z-score方法检测异常值
|
|
z_scores = np.abs((data - mean) / std)
|
|
outlier_indices = np.where(z_scores > threshold)[0].tolist()
|
|
|
|
return outlier_indices
|
|
|
|
except Exception as e:
|
|
logger.error(f'异常值检测失败: {e}')
|
|
return []
|
|
|
|
def _generate_charts(self, data: Dict[str, List], session_id: str) -> Dict[str, str]:
|
|
"""生成图表"""
|
|
try:
|
|
charts = {}
|
|
|
|
# 创建会话专用目录
|
|
session_charts_dir = self.charts_dir / session_id
|
|
session_charts_dir.mkdir(exist_ok=True)
|
|
|
|
# 生成平衡指数时间序列图
|
|
if data['timestamps'] and data['balance_index']:
|
|
chart_path = self._create_balance_chart(data, session_charts_dir)
|
|
if chart_path:
|
|
charts['balance_chart'] = str(chart_path)
|
|
|
|
# 生成压力分布图
|
|
if data['timestamps'] and data['pressure_left'] and data['pressure_right']:
|
|
chart_path = self._create_pressure_chart(data, session_charts_dir)
|
|
if chart_path:
|
|
charts['pressure_chart'] = str(chart_path)
|
|
|
|
# 生成重心轨迹图
|
|
if data['center_of_pressure_x'] and data['center_of_pressure_y']:
|
|
chart_path = self._create_cop_trajectory_chart(data, session_charts_dir)
|
|
if chart_path:
|
|
charts['cop_trajectory_chart'] = str(chart_path)
|
|
|
|
# 生成姿态角度图
|
|
if data['timestamps'] and data['imu_pitch'] and data['imu_roll']:
|
|
chart_path = self._create_posture_chart(data, session_charts_dir)
|
|
if chart_path:
|
|
charts['posture_chart'] = str(chart_path)
|
|
|
|
return charts
|
|
|
|
except Exception as e:
|
|
logger.error(f'图表生成失败: {e}')
|
|
return {}
|
|
|
|
def _create_balance_chart(self, data: Dict[str, List], output_dir: Path) -> Optional[Path]:
|
|
"""创建平衡指数图表"""
|
|
try:
|
|
plt.figure(figsize=(12, 6))
|
|
|
|
timestamps = data['timestamps'][:len(data['balance_index'])]
|
|
plt.plot(timestamps, data['balance_index'], 'b-', linewidth=1.5, label='平衡指数')
|
|
|
|
# 添加平均线
|
|
mean_balance = np.mean(data['balance_index'])
|
|
plt.axhline(y=mean_balance, color='r', linestyle='--', alpha=0.7, label=f'平均值: {mean_balance:.3f}')
|
|
|
|
plt.title('平衡指数时间序列', fontsize=14, fontweight='bold')
|
|
plt.xlabel('时间', fontsize=12)
|
|
plt.ylabel('平衡指数', fontsize=12)
|
|
plt.legend()
|
|
plt.grid(True, alpha=0.3)
|
|
|
|
# 格式化时间轴
|
|
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))
|
|
plt.xticks(rotation=45)
|
|
|
|
plt.tight_layout()
|
|
|
|
chart_path = output_dir / 'balance_chart.png'
|
|
plt.savefig(chart_path, dpi=300, bbox_inches='tight')
|
|
plt.close()
|
|
|
|
return chart_path
|
|
|
|
except Exception as e:
|
|
logger.error(f'平衡图表创建失败: {e}')
|
|
plt.close()
|
|
return None
|
|
|
|
def _create_pressure_chart(self, data: Dict[str, List], output_dir: Path) -> Optional[Path]:
|
|
"""创建压力分布图表"""
|
|
try:
|
|
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10))
|
|
|
|
timestamps = data['timestamps'][:min(len(data['pressure_left']), len(data['pressure_right']))]
|
|
pressure_left = data['pressure_left'][:len(timestamps)]
|
|
pressure_right = data['pressure_right'][:len(timestamps)]
|
|
|
|
# 上图：左右脚压力对比
|
|
ax1.plot(timestamps, pressure_left, 'b-', linewidth=1.5, label='左脚压力')
|
|
ax1.plot(timestamps, pressure_right, 'r-', linewidth=1.5, label='右脚压力')
|
|
ax1.set_title('左右脚压力对比', fontsize=14, fontweight='bold')
|
|
ax1.set_ylabel('压力值', fontsize=12)
|
|
ax1.legend()
|
|
ax1.grid(True, alpha=0.3)
|
|
|
|
# 下图：压力比例
|
|
total_pressure = np.array(pressure_left) + np.array(pressure_right)
|
|
left_ratio = np.array(pressure_left) / total_pressure * 100
|
|
right_ratio = np.array(pressure_right) / total_pressure * 100
|
|
|
|
ax2.plot(timestamps, left_ratio, 'b-', linewidth=1.5, label='左脚比例')
|
|
ax2.plot(timestamps, right_ratio, 'r-', linewidth=1.5, label='右脚比例')
|
|
ax2.axhline(y=50, color='g', linestyle='--', alpha=0.7, label='理想平衡线')
|
|
ax2.set_title('左右脚压力比例', fontsize=14, fontweight='bold')
|
|
ax2.set_xlabel('时间', fontsize=12)
|
|
ax2.set_ylabel('压力比例 (%)', fontsize=12)
|
|
ax2.legend()
|
|
ax2.grid(True, alpha=0.3)
|
|
|
|
# 格式化时间轴
|
|
for ax in [ax1, ax2]:
|
|
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))
|
|
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)
|
|
|
|
plt.tight_layout()
|
|
|
|
chart_path = output_dir / 'pressure_chart.png'
|
|
plt.savefig(chart_path, dpi=300, bbox_inches='tight')
|
|
plt.close()
|
|
|
|
return chart_path
|
|
|
|
except Exception as e:
|
|
logger.error(f'压力图表创建失败: {e}')
|
|
plt.close()
|
|
return None
|
|
|
|
def _create_cop_trajectory_chart(self, data: Dict[str, List], output_dir: Path) -> Optional[Path]:
|
|
"""创建重心轨迹图表"""
|
|
try:
|
|
plt.figure(figsize=(10, 8))
|
|
|
|
x_positions = data['center_of_pressure_x']
|
|
y_positions = data['center_of_pressure_y']
|
|
|
|
# 绘制轨迹
|
|
plt.plot(x_positions, y_positions, 'b-', linewidth=1, alpha=0.7, label='重心轨迹')
|
|
plt.scatter(x_positions[0], y_positions[0], color='g', s=100, marker='o', label='起始点')
|
|
plt.scatter(x_positions[-1], y_positions[-1], color='r', s=100, marker='s', label='结束点')
|
|
|
|
# 添加中心点
|
|
center_x = np.mean(x_positions)
|
|
center_y = np.mean(y_positions)
|
|
plt.scatter(center_x, center_y, color='orange', s=150, marker='*', label='平均中心')
|
|
|
|
# 添加置信椭圆
|
|
self._add_confidence_ellipse(x_positions, y_positions, plt.gca())
|
|
|
|
plt.title('重心轨迹图', fontsize=14, fontweight='bold')
|
|
plt.xlabel('X方向位移 (mm)', fontsize=12)
|
|
plt.ylabel('Y方向位移 (mm)', fontsize=12)
|
|
plt.legend()
|
|
plt.grid(True, alpha=0.3)
|
|
plt.axis('equal')
|
|
|
|
plt.tight_layout()
|
|
|
|
chart_path = output_dir / 'cop_trajectory_chart.png'
|
|
plt.savefig(chart_path, dpi=300, bbox_inches='tight')
|
|
plt.close()
|
|
|
|
return chart_path
|
|
|
|
except Exception as e:
|
|
logger.error(f'重心轨迹图表创建失败: {e}')
|
|
plt.close()
|
|
return None
|
|
|
|
def _add_confidence_ellipse(self, x: List[float], y: List[float], ax, n_std: float = 2.0):
|
|
"""添加置信椭圆"""
|
|
try:
|
|
from matplotlib.patches import Ellipse
|
|
import matplotlib.transforms as transforms
|
|
|
|
x_arr = np.array(x)
|
|
y_arr = np.array(y)
|
|
|
|
cov = np.cov(x_arr, y_arr)
|
|
pearson = cov[0, 1] / np.sqrt(cov[0, 0] * cov[1, 1])
|
|
|
|
ell_radius_x = np.sqrt(1 + pearson)
|
|
ell_radius_y = np.sqrt(1 - pearson)
|
|
|
|
ellipse = Ellipse((0, 0), width=ell_radius_x * 2, height=ell_radius_y * 2,
|
|
facecolor='none', edgecolor='red', alpha=0.5, linestyle='--')
|
|
|
|
scale_x = np.sqrt(cov[0, 0]) * n_std
|
|
scale_y = np.sqrt(cov[1, 1]) * n_std
|
|
|
|
mean_x = np.mean(x_arr)
|
|
mean_y = np.mean(y_arr)
|
|
|
|
transf = transforms.Affine2D().scale(scale_x, scale_y).translate(mean_x, mean_y)
|
|
ellipse.set_transform(transf + ax.transData)
|
|
|
|
ax.add_patch(ellipse)
|
|
|
|
except Exception as e:
|
|
logger.warning(f'置信椭圆添加失败: {e}')
|
|
|
|
def _create_posture_chart(self, data: Dict[str, List], output_dir: Path) -> Optional[Path]:
|
|
"""创建姿态角度图表"""
|
|
try:
|
|
plt.figure(figsize=(12, 8))
|
|
|
|
timestamps = data['timestamps'][:min(len(data['imu_pitch']), len(data['imu_roll']))]
|
|
pitch = data['imu_pitch'][:len(timestamps)]
|
|
roll = data['imu_roll'][:len(timestamps)]
|
|
|
|
plt.subplot(2, 1, 1)
|
|
plt.plot(timestamps, pitch, 'b-', linewidth=1.5, label='俯仰角')
|
|
plt.axhline(y=0, color='g', linestyle='--', alpha=0.7, label='理想位置')
|
|
plt.title('俯仰角变化', fontsize=14, fontweight='bold')
|
|
plt.ylabel('角度 (度)', fontsize=12)
|
|
plt.legend()
|
|
plt.grid(True, alpha=0.3)
|
|
|
|
plt.subplot(2, 1, 2)
|
|
plt.plot(timestamps, roll, 'r-', linewidth=1.5, label='翻滚角')
|
|
plt.axhline(y=0, color='g', linestyle='--', alpha=0.7, label='理想位置')
|
|
plt.title('翻滚角变化', fontsize=14, fontweight='bold')
|
|
plt.xlabel('时间', fontsize=12)
|
|
plt.ylabel('角度 (度)', fontsize=12)
|
|
plt.legend()
|
|
plt.grid(True, alpha=0.3)
|
|
|
|
# 格式化时间轴
|
|
for i in range(1, 3):
|
|
ax = plt.subplot(2, 1, i)
|
|
ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))
|
|
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)
|
|
|
|
plt.tight_layout()
|
|
|
|
chart_path = output_dir / 'posture_chart.png'
|
|
plt.savefig(chart_path, dpi=300, bbox_inches='tight')
|
|
plt.close()
|
|
|
|
return chart_path
|
|
|
|
except Exception as e:
|
|
logger.error(f'姿态图表创建失败: {e}')
|
|
plt.close()
|
|
return None
|
|
|
|
def _generate_summary(self, statistical: Dict, trends: Dict, anomalies: Dict) -> Dict[str, Any]:
|
|
"""生成分析摘要"""
|
|
try:
|
|
summary = {
|
|
'overall_assessment': 'normal',
|
|
'key_findings': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
# 分析平衡状况
|
|
if 'balance' in statistical:
|
|
balance_mean = statistical['balance']['mean']
|
|
if balance_mean > 0.3:
|
|
summary['key_findings'].append('平衡能力较差，重心摆动较大')
|
|
summary['recommendations'].append('建议进行平衡训练，加强核心肌群锻炼')
|
|
summary['overall_assessment'] = 'poor'
|
|
elif balance_mean > 0.15:
|
|
summary['key_findings'].append('平衡能力一般，有改善空间')
|
|
summary['recommendations'].append('建议适当增加平衡练习')
|
|
if summary['overall_assessment'] == 'normal':
|
|
summary['overall_assessment'] = 'fair'
|
|
else:
|
|
summary['key_findings'].append('平衡能力良好')
|
|
|
|
# 分析压力分布
|
|
if 'pressure_distribution' in statistical:
|
|
asymmetry = statistical['pressure_distribution']['asymmetry']
|
|
if asymmetry > 0.2:
|
|
summary['key_findings'].append('左右脚压力分布不均，存在明显偏重')
|
|
summary['recommendations'].append('注意纠正站立姿势，均匀分配体重')
|
|
if summary['overall_assessment'] in ['normal', 'fair']:
|
|
summary['overall_assessment'] = 'fair'
|
|
|
|
# 分析姿态稳定性
|
|
if 'posture' in statistical:
|
|
posture_stability = statistical['posture'].get('posture_stability', 1.0)
|
|
if posture_stability < 0.7:
|
|
summary['key_findings'].append('姿态稳定性较差，身体摆动较大')
|
|
summary['recommendations'].append('建议进行姿态矫正训练')
|
|
if summary['overall_assessment'] == 'normal':
|
|
summary['overall_assessment'] = 'fair'
|
|
|
|
# 分析异常情况
|
|
if anomalies:
|
|
total_anomalies = 0
|
|
for key, value in anomalies.items():
|
|
if isinstance(value, dict) and 'count' in value:
|
|
total_anomalies += value['count']
|
|
|
|
if total_anomalies > 10:
|
|
summary['key_findings'].append(f'检测到{total_anomalies}个异常数据点')
|
|
summary['recommendations'].append('建议重新进行检测或咨询专业医师')
|
|
|
|
# 默认建议
|
|
if not summary['recommendations']:
|
|
summary['recommendations'].append('继续保持良好的平衡训练习惯')
|
|
|
|
return summary
|
|
|
|
except Exception as e:
|
|
logger.error(f'摘要生成失败: {e}')
|
|
return {
|
|
'overall_assessment': 'unknown',
|
|
'key_findings': ['分析过程中出现错误'],
|
|
'recommendations': ['建议重新进行检测']
|
|
}
|
|
|
|
def generate_report(self, session_id: str) -> str:
|
|
"""生成PDF报告"""
|
|
try:
|
|
# 这里应该从数据库获取会话数据和分析结果
|
|
# 目前返回一个模拟的报告路径
|
|
report_path = self.export_dir / f'report_{session_id}_{datetime.now().strftime("%Y%m%d_%H%M%S")}.pdf'
|
|
|
|
# 创建简单的PDF报告
|
|
self._create_simple_pdf_report(session_id, report_path)
|
|
|
|
return str(report_path)
|
|
|
|
except Exception as e:
|
|
logger.error(f'报告生成失败: {e}')
|
|
raise
|
|
|
|
def _create_simple_pdf_report(self, session_id: str, output_path: Path):
|
|
"""创建简单的PDF报告"""
|
|
try:
|
|
from reportlab.pdfgen import canvas
|
|
from reportlab.lib.pagesizes import A4
|
|
|
|
c = canvas.Canvas(str(output_path), pagesize=A4)
|
|
width, height = A4
|
|
|
|
# 标题
|
|
c.setFont("Helvetica-Bold", 20)
|
|
c.drawString(50, height - 50, "Balance Detection Report")
|
|
|
|
# 会话信息
|
|
c.setFont("Helvetica", 12)
|
|
y_position = height - 100
|
|
|
|
c.drawString(50, y_position, f"Session ID: {session_id}")
|
|
y_position -= 20
|
|
c.drawString(50, y_position, f"Report Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
y_position -= 40
|
|
|
|
# 报告内容
|
|
c.drawString(50, y_position, "Analysis Summary:")
|
|
y_position -= 20
|
|
c.drawString(70, y_position, "• Balance assessment completed")
|
|
y_position -= 20
|
|
c.drawString(70, y_position, "• Posture analysis performed")
|
|
y_position -= 20
|
|
c.drawString(70, y_position, "• Movement patterns evaluated")
|
|
|
|
c.save()
|
|
logger.info(f'PDF报告已生成: {output_path}')
|
|
|
|
except Exception as e:
|
|
logger.error(f'PDF报告创建失败: {e}')
|
|
raise
|
|
|
|
def export_data(self, session_id: str, format: str = 'csv') -> str:
|
|
"""导出数据"""
|
|
try:
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
if format.lower() == 'csv':
|
|
export_path = self.export_dir / f'data_{session_id}_{timestamp}.csv'
|
|
# 这里应该从数据库获取数据并导出为CSV
|
|
# 目前创建一个示例文件
|
|
with open(export_path, 'w', encoding='utf-8') as f:
|
|
f.write('timestamp,data_type,value\n')
|
|
f.write(f'{datetime.now().isoformat()},sample,0.5\n')
|
|
|
|
elif format.lower() == 'json':
|
|
export_path = self.export_dir / f'data_{session_id}_{timestamp}.json'
|
|
# 导出为JSON格式
|
|
sample_data = {
|
|
'session_id': session_id,
|
|
'export_time': datetime.now().isoformat(),
|
|
'data': []
|
|
}
|
|
with open(export_path, 'w', encoding='utf-8') as f:
|
|
json.dump(sample_data, f, ensure_ascii=False, indent=2)
|
|
|
|
else:
|
|
raise ValueError(f'不支持的导出格式: {format}')
|
|
|
|
logger.info(f'数据已导出: {export_path}')
|
|
return str(export_path)
|
|
|
|
except Exception as e:
|
|
logger.error(f'数据导出失败: {e}')
|
|
raise |