Self-Define/api/self_define_api.py

from fastapi import FastAPI, HTTPException, Body
from pydantic import BaseModel
import numpy as np
import pandas as pd
import json
import uvicorn
from typing import Optional, List


app = FastAPI()

class DataRequest(BaseModel):
    
    data: List
    start_index: Optional[int] = None
    end_index: Optional[int] = None

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

@app.post("/CalculateFeatures/")
async def calculate_features(request: DataRequest = Body(...)):
    try:
        json_data = request.model_dump()
        result = load_and_calculate_features(json_data)
        return result
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8666)