# %% import numpy as np import pandas as pd import json with open("input.json",'r') as f: json_data = json.load(f) json_data # %% # 定义加载数据并计算特征值的函数 def load_and_calculate_features(json_data): # 定义计算统计指标的函数 def _calculate_stats(data): # 确保数据是数值类型 data = pd.to_numeric(data, errors='coerce') mean_value = np.mean(data) max_value = np.max(data) min_value = np.min(data) Xrms = np.sqrt(np.mean(data ** 2)) # 加速度均方根值(有效值) Xp = (max_value - min_value) / 2 # 峰值(单峰最大值) Xpp = max_value - min_value # 峰峰值 Cf = Xp / Xrms # 峰值指标 Sk = np.mean((data - mean_value) ** 3) / Xrms ** 3 # 歪度指标 Sf = Xrms / mean_value # 波形指标 If = Xp / np.mean(np.abs(data)) # 脉冲指标 Xr = np.mean(np.sqrt(np.abs(data))) ** 2 # 方根幅值 Ce = Xp / Xr # 裕度指标 return { 'Xrms': Xrms, 'Mean': mean_value, 'Max': max_value, 'Min': min_value, 'Xp': Xp, 'Xpp': Xpp, 'Cf': Cf, 'Sk': Sk, 'Ce': Ce, 'If': If, 'Sf': Sf, } if json_data['start_index'] is not None and json_data['end_index'] is not None: df = pd.DataFrame(json_data['data']) start_index = json_data['start_index'] end_index = json_data['end_index'] selected_data = df[(df['index'] >= start_index) & (df['index'] <= end_index)] stats = _calculate_stats(selected_data['value']) for feature, value in stats.items(): df[feature] = np.where((df['index'] >= start_index) & (df['index'] <= end_index), value, 0) json_output = df.to_json(date_format='iso', orient='records', force_ascii=False) else: df = pd.DataFrame(json_data['data']) stats = _calculate_stats(df['value']) for feature, value in stats.items(): df[feature] = value json_output = df.to_json(date_format='iso', orient='records', force_ascii=False) return json_output # load_and_calculate_features(json_data)