CMS/api_20241201.py

# coding=utf-8
from fastapi import FastAPI,  UploadFile, HTTPException
from fastapi.responses import JSONResponse
from typing import List, Optional
from pydantic import BaseModel
import os
import json
import shutil
import pandas as pd
import uvicorn
from cms_class_20241201 import CMSAnalyst
from typing import List, Optional, Union
from pydantic import BaseModel, Field, model_validator

# 初始化 FastAPI 应用
app = FastAPI()


'''
input:
{
    "ids":[12345,121212],
    "windCode":"xxxx",
    "analysisType":"xxxxx",
    "fmin":int(xxxx) (None),
    "fmax":"int(xxxx) (None),
}


output:

{
data:[
    {
    "type":"envelope_spectrum",
    "x":list(x),
    "y":list(y),
    "title":title,
    "xaxis":xaxis,
    "yaxis":yaxis
    },
    {
    "type":"bandpass_filtered",
    "x":list(x),
    "y":list(y),
    "title":title,
    "xaxis":xaxis,
    "yaxis":yaxis
    },


    # 其他情况：
    {
    "type":"trend_analysis",
    "Mean": mean_value,
    "Max": max_value,
    "Min": min_value,
    "Xrms": Xrms,
    "Xp": Xp,
    "If": If,
    "Cf": Cf,
    "Sf": Sf,
    "Ce": Ce,
    "Cw": Cw,
    "Cq": Cq
    }
],

"code": 200,
"message":success
}

'''
# 请求模型定义
class AnalysisInput(BaseModel):
    # ids: List
    ids: List[int]
    windCode: str
    analysisType: str
    fmin: Optional[int] = None
    fmax: Optional[int] = None

    @model_validator(mode='before')
    def convert_ids(cls, values):
        if isinstance(values.get('ids'), int):
            values['ids'] = [values['ids']]
        return values

# url = "http://127.0.0.1:8888/analysis/trend"


@app.post("/analysis/{analysisType}")
async def analyze(analysisType: str, input_data:AnalysisInput):

    analysis_map = {
        "envelope": "envelope_spectrum",#包络谱分析
        "frequency": "frequency_domain",#频域分析
        "time": "time_domain",#时域分析
        "trend": "trend_analysis"#趋势分析
    }


    if analysisType not in analysis_map:
        raise HTTPException(status_code=400, detail="非可用的分析类型")


    try:
        cms = CMSAnalyst(input_data.fmin, input_data.fmax, input_data.windCode, input_data.ids)

        func = getattr(cms, analysis_map[analysisType])
        # if callable(func):
        #     func_res = func()
        # func_res = json.loads(func_res)
        # func_res['type'] = analysisType
        # # return {
        # #     "message": "success",
        # #     "data": [func_res],
        # # }
        if callable(func):    # 用于判断一个对象是否可以被调用,是的话返回true
            func_res = func()
        if isinstance(func_res, str):
            func_res = json.loads(func_res)    # 字符串转化为字典形式
        if isinstance(func_res, dict):
            func_res['type'] = analysisType
        elif isinstance(func_res, list):
            func_res = {'type': analysisType, 'data': func_res}
        else:
            # 处理其他情况，例如其他数据类型
            func_res = {'type': analysisType, 'data': str(func_res)}
        return json.dumps(func_res,ensure_ascii=False)   #对象转化为字符串形式
        # return JSONResponse(content=func_res)#返回json格式
    except Exception as e:
          return {"message": "error", "detail": str(e)}

# @app.post("/analysis/{analysisType}")
# async def analyze(analysisType: str, input_data:AnalysisInput):
#     try:
#         analysis_map = {
#             "envelope": ["envelope_spectrum","bandpass_filtered"],
#             "frequency": "frequency_domain",
#             "time": "time_domain",
#             "trend": "trend_analysis"
#         }


#         if analysisType not in analysis_map:
#             raise HTTPException(status_code=400, detail="非可用的分析类型")


#         try:
#             cms = CMSAnalyst(input_data.fmin, input_data.fmax, input_data.windCode, input_data.ids)
            
#             if analysisType == "envelope":
#                 tmp = []
#                 for type_func in analysis_map[analysisType]:
#                     func = getattr(cms, type_func)
#                     if callable(func):
#                         func_res = func()
#                     func_res['type'] = analysisType
#                     tmp.append(func_res)

#                 return {
#                     "message": "success",
#                     "data": tmp,
#                 }
            
#             else:
#                 func = getattr(cms, analysis_map[analysisType])
#                 if callable(func):
#                     func_res = func()
#                 func_res['type'] = analysisType
#                 return {
#                     "message": "success",
#                     "data": [func_res],
#                 }
#         except Exception as e:
#             return {"message": "error", "detail": str(e)}
#     except:
#         raise HTTPException(status_code=400, detail="接口返回错误")


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