model-lab/app.py

# -*- coding: utf-8 -*-
import json
import traceback

import PCA_Test
import PCA_Test_offline
import numpy as np
from flask import Flask
from flask import request

app = Flask(__name__)


# app.permanent_session_lifetime = timedelta(minutes=5)

@app.route('/', methods=["POST", "GET"])
def FirstPage():
    return "This is a Flask WebAPi"


@app.route('/api/test/Test_offline', methods=["GET"])
def GetTest_offline():
    try:
        result = ""
        jsonstr = json.loads(request.args.get("json"))
        model_id = jsonstr["Model_id"]
        version = jsonstr["version"]
        res = PCA_Test_offline.get_model_by_id_and_version(model_id, version)
        filename = res["algorithm"]
        Data = jsonstr["Test_Data"]
        if filename == "PCA":
            model = res["para"]["Model_info"]
            lock = []
            point_info = res["pointInfo"]
            for i in range(len(point_info)):
                try:
                    if point_info[i]["lock"]:
                        lock.append(i)
                except:
                    continue
            result = PCA_Test_offline.pca(model, lock, np.array(Data))
        elif filename == "sPCA":
            model = res["para"]["Model_info"]
            lock = []
            point_info = res["pointInfo"]
            for i in range(len(point_info)):
                try:
                    if point_info[i]["lock"]:
                        lock.append(i)
                except:
                    continue
            import sPCA
            result = sPCA.test_offline(model, lock, np.array(Data))
            return json.dumps(result)
        else:
            import ANN_Test_offline
            info = res["para"]
            result = ANN_Test_offline.test_offline_main(info["mms1"], info["mms2"], Data, jsonstr["Target_Data"],
                                                        info["filename"], info["weight"])
            return json.dumps(result)
    except Exception as e:
        result = json.dumps({"msg": traceback.format_exc()}, ensure_ascii=False)
    return result


@app.route('/api/predict', methods=["POST"])
def get_predict():
    try:
        result = ""
        jsonstr = json.loads(request.data)
        model_id = jsonstr["Model_id"]
        version = jsonstr["version"]
        res = PCA_Test_offline.get_model_by_id_and_version(model_id, version)
        filename = res["algorithm"]
        Data = jsonstr["Test_Data"]
        if filename == "PCA":
            model = res["para"]["Model_info"]
            lock = []
            point_info = res["pointInfo"]
            for i in range(len(point_info)):
                try:
                    if point_info[i]["lock"]:
                        lock.append(i)
                except:
                    continue
            result = PCA_Test_offline.pca(model, lock, np.array(Data))
        elif filename == "GMM_O":
            # model = res["para"]["Model_info"]
            import GMM_PCA_test_offline
            result = GMM_PCA_test_offline.main(jsonstr)
            return json.dumps(result)
        else:
            import ANN_Test_offline
            info = res["para"]
            result = ANN_Test_offline.test_offline_main(info["mms1"], info["mms2"], Data, jsonstr["Target_Data"],
                                                        info["filename"], info["weight"])
            return json.dumps(result)
    except Exception as e:
        result = json.dumps({"msg": traceback.format_exc()}, ensure_ascii=False)
    return result


@app.route('/api/test/ANN_Test_offline', methods=["GET", "POST"])
def GetANNTestOffline():
    try:
        info = json.loads(request.data)
        import ANN_Test_offline
        result = ANN_Test_offline.main(info["mms1"], info["mms2"], np.array(info["testDataX"]),
                                       np.array(info["testDataY"]), info["filename"], info["weight"])
    except Exception as e:
        result = {"error_msg": str(e)}
    return json.dumps(result)


@app.route('/api/test/Train', methods=["POST", "GET"])
def GetTrain():
    result = ""
    jsonstr = json.loads(request.args.get("json"))
    filename = request.args.get("type")
    if filename == "PCA":
        import PCA_Train
        result = PCA_Train.main(jsonstr)
    elif filename == "GMM":
        import GMM_train
        result = GMM_train.main(jsonstr)
    return result


@app.route('/api/test/ASSESS', methods=["POST", "GET"])
def GetAssess():
    result = {}
    try:
        import ASSESS
        HSS = ASSESS.HealthyScoringSystem()
        HSS.getInputDate()
        HSS.Scoring()
        HSS.Conditioning()
        HSS.weightedSumScoring('subsystem')
        HSS.weightedSumScoring('system')
        HSS.weightedSumScoring('unit')
        HSS.outputScore()
        return result
    except Exception as e:
        return json.dumps({"error_msg": traceback.format_exc()})


@app.route('/api/test/ClearTrain', methods=["POST", "GET"])
def GetClearTrain():
    try:
        result = ""
        jsonstr = json.loads(request.data)
        filename = jsonstr["type"]
        if filename == "PCA":
            import PCA_Train
            result = PCA_Train.clearmain(jsonstr)
        elif filename == "AANN":  # AANN added by rsj 2021-5-3
            import AANN_Train
            result = AANN_Train.clearmain(jsonstr)
        elif filename == "sPCA":
            import sPCA
            result = sPCA.clearmain(jsonstr)
        return result
    except Exception as e:
        return json.dumps({"error_msg": traceback.format_exc()})


@app.route('/api/test/Test', methods=["GET", "POST"])
def GetTest():
    try:
        result = ""
        info = json.loads(request.data)
        model_id = info["Model_id"]
        version = info["version"]
        if version == "v-test":
            res = PCA_Test.get_model_by_ID(model_id, version)
        else:
            res = PCA_Test.get_model_by_id_and_version(model_id, version)
        filename = res["algorithm"]
        if filename == "PCA":
            Test_Data = info["Test_Data"]
            points = Test_Data["points"]
            time1 = Test_Data["time"]
            interval = Test_Data["interval"]
            model = res["para"]["Model_info"]
            Data = PCA_Test.get_history_value(points, time1, interval)
            result = PCA_Test.pca(model, Data)
            index = time1.index(",")
            result["time"] = time1[:index:]
        elif filename == "GMM_O":
            import GMM_PCA_test
            result = GMM_PCA_test.main(info)
            Test_Data = info["Test_Data"]
            time1 = Test_Data["time"]
            index = time1.index(",")
            result["time"] = time1[:index:]
        elif filename == "AANN":
            import AANN_Fit
            Test_Data = info["Test_Data"]
            points = Test_Data["points"]
            time1 = Test_Data["time"]
            interval = Test_Data["interval"]
            # print('1111111',res["para"])
            Data = AANN_Fit.get_history_value(points, time1, interval)
            model = res["para"]["Model_info"]
            result = AANN_Fit.AANN_Test(model, Data)
            index = time1.index(",")
            result["time"] = time1[:index:]

        else:
            pass
        return json.dumps(result)
    except Exception as e:
        return json.dumps({"error_msg": traceback.format_exc()})


@app.route("/api/aann_clean_test", methods=["POST"])
def get_aann_clean_test():
    try:
        info = json.loads(request.data)
        import AANN_Fit
        result = AANN_Fit.clean_main(info)
    except Exception as e:
        result = {"error_msg": str(e)}
    return json.dumps(result)


@app.route('/api/test/ANN_Train_offline', methods=["GET", "POST"])
def GetANNTrainOffline():
    try:
        info = json.loads(request.data)
        import ANN_Train_offline
        result = ANN_Train_offline.TrainOffline(np.array(info["Train_Data_X"]), np.array(info["Train_Data_Y"]),
                                                info["hide"], eval(info["iter"]))
        return result
    except Exception as e:
        result = {"error_msg": str(e)}
        return json.dumps(result)


@app.route('/api/test/ANN_Train', methods=["POST"])
def GetANNTrain():
    try:
        info = json.loads(request.data)
        import ANN_Train_offline
        result = ANN_Train_offline.clearmain(info)
        return json.dumps(result)
    except Exception as e:
        result = {"error": str(e)}
        return json.dumps(result)


@app.route('/api/test/ANN_Test', methods=["POST"])
def GetANNTest():
    try:
        info = json.loads(request.data)
        points = info['point']
        time1 = info["time"]
        datatype = info['type']
        interval = info['interval']
        import ANN_Test_offline
        data_x, origndata = ANN_Test_offline.get_history_value(points, time1, interval, datatype)
        res = json.loads(info["model"])
        weight = res["weight"]
        filename = res["filename"]
        mms1 = res["mms1"]
        mms2 = res["mms2"]
        d_type = list(enumerate(datatype.split(",")))
        output_data = [origndata.tolist()[item[0]] for item in d_type if item[1] == "1"]
        result = ANN_Test_offline.main(mms1, mms2, data_x, np.array(output_data).T, filename, weight)
        result["sampleData"] = origndata.tolist()
    except Exception as e:
        result = {}
        result["error_msg"] = str(e)
    return json.dumps(result)


@app.route('/api/test/PLS_Train_online', methods=["POST", "GET"])
def getPLSTrainOnline():
    result = ""
    jsonstr = json.loads(request.data)
    import PLS_Train
    result = PLS_Train.main(jsonstr)
    return result


@app.route('/api/test/PLS_Test_online', methods=["POST", "GET"])
def GetPLSTestOnline():
    result = ""
    info = json.loads(request.data)
    import PLS_Test
    result = PLS_Test.main(info)
    return result


@app.route('/api/test/getliner', methods=["POST", "GET"])
def get_liner():
    data = json.loads(request.data)
    import Liner
    degree = data["type"]
    inputdata = np.array(data["Samples"])
    x = inputdata[:, 0].reshape(-1, 1)
    y = inputdata[:, 1].reshape(-1, 1)
    return Liner.main(degree, x, y)


@app.route('/HeatRate', methods=["POST", "GET"])
def get_heat_rate():
    data = json.loads(request.args.get("json"))
    import heatrate
    result = heatrate.main(data)
    return result


@app.route('/api/Test_offlinerb', methods=["POST"])
def get_test_offlinerb():
    info = json.loads(request.data)
    model_id = info["Model_id"]
    version = info["version"]
    if version == "v-test":
        res = PCA_Test.get_model_by_ID(model_id, version)
    else:
        res = PCA_Test.get_model_by_id_and_version(model_id, version)
    filename = res["algorithm"]
    if filename == "PCA":
        lock = []
        point_info = res["pointInfo"]
        for i in range(len(point_info)):
            try:
                if point_info[i]["lock"]:
                    lock.append(i)
            except:
                continue
        import pca_test_by_rb_plot
        Test_Data = info["Test_Data"]
        points = Test_Data["points"]
        time1 = Test_Data["time"]
        interval = Test_Data["interval"]
        model = res["para"]["Model_info"]
        Data = pca_test_by_rb_plot.get_history_value(points, time1, interval)
        # from smote import smote_new
        # Data = smote_new(Data)
        result = pca_test_by_rb_plot.pca(model, lock, Data)
        index = time1.index(",")
        result["time"] = time1[:index:]
    return json.dumps(result)


@app.route("/api/pca_off_train", methods=["POST"])
def get_pca_off_train():
    try:
        info = json.loads(request.data)
        from pca_train_off import cleanmain
        result = cleanmain(info)
    except Exception as e:
        result = json.dumps({"error_msg": traceback.format_exc()})
    return result


@app.route('/api/pcamtcltest', methods=["POST"])
def get_test_pcamtcltest():
    try:
        info = json.loads(request.data)
        import pcamtcltest
        result = pcamtcltest.cleanmain(info)
        return json.dumps(result)
    except Exception as e:
        result = {"error_msg": str(e)}
        return json.dumps(result)


@app.route('/api/pcamtcltest_recon', methods=["POST"])
def get_test_pcamtcltest_recon():
    try:
        info = json.loads(request.data)
        import pcamtcltest_recon
        result = pcamtcltest_recon.cleanmain(info)
        return json.dumps(result)
    except Exception as e:
        result = {"error_msg": str(e)}
        return json.dumps(result)


@app.route("/api/mtcltest_recon_exa", methods=["POST"])
def get_mtcltest_recon_exa():
    try:
        info = json.loads(request.data)
        filename = info["Model_alg"]
        if filename == "PCA":
            import pcamtcltest_recon
            result = pcamtcltest_recon.main(info)
            return json.dumps(result)
        elif filename == "AANN":
            import aannmtcltest_recon
            result = aannmtcltest_recon.main(info)
            return json.dumps(result)
    except Exception as e:
        result = {"error_msg": str(e)}
        return json.dumps(result)


@app.route("/api/ann_clean_test", methods=["POST"])
def get_ann_clean_test():
    try:
        info = json.loads(request.data)
        import ANN_Test_offline
        result = ANN_Test_offline.clean_main(info)
    except Exception as e:
        result = {"error_msg": str(e)}
    return json.dumps(result)


@app.route("/api/test/model_performance", methods=["POST"])
def get_model_performance():
    info = json.loads(request.data)
    import model_performance
    result = model_performance.main(info)
    return json.dumps(result).encode("utf-8").decode("unicode_escape")


@app.route("/api/model_performance_data", methods=["POST"])
def get_model_performance_data():
    info = json.loads(request.data)
    import model_performance
    result = model_performance.data_main(info)
    return json.dumps(result).encode("utf-8").decode("unicode_escape")


@app.route("/api/test/spca_train", methods=["POST"])
def get_spca_train():
    try:
        info = json.loads(request.data)
        import spca
        result = spca.main(info)
    except Exception as e:
        result = json.dumps({"error_msg": str(e)})
    return result


@app.route("/api/test/spca_test", methods=["POST"])
def get_spca_test():
    try:
        info = json.loads(request.data)
        import spca
        result = spca.test_main(info)
    except Exception as e:
        result = json.dumps({"error_msg": traceback.format_exc()})
    return result


@app.route("/api/test/clean_data", methods=["POST"])
def get_clean_data():
    try:
        info = json.loads(request.data)
        import clean
        result = clean.main(info)
    except Exception as e:
        result = json.dumps({"error_msg": traceback.format_exc()})
    return result


@app.route("/api/model_coverage", methods=["POST"])
def model_coverage():
    try:
        info = json.loads(request.data)
        import model_coverage
        result = model_coverage.main(info)
    except Exception as e:
        result = json.dumps({"error_msg": traceback.format_exc()})
    return result


@app.route('/api/get_smote_data', methods=["POST"])
def get_smote_data():
    try:
        info = json.loads(request.data)
        from smote import smote_data
        res = smote_data(info)
    except Exception as e:
        res = {"error_msg": traceback.format_exc()}
    return json.dumps(res)


@app.route('/api/ae_train', methods=["POST"])
def ae_train():
    info = json.loads(request.data)
    from sae_main import train_main
    res = train_main(info)
    return res


@app.route('/api/ae_test', methods=["POST"])
def ae_test():
    try:
        info = json.loads(request.data)
        from sae_main import test_main
        res = test_main(info)
    except Exception as e:
        res = json.dumps({"error_msg": traceback.format_exc()})
    return res


@app.route("/api/get_pca_cp_mtcl_test", methods=["POST"])
def get_pca_cp_mtcl_test():
    try:
        info = json.loads(request.data)
        from pca import cp_main  ##
        res = cp_main(info)
    except Exception as e:
        res = json.dumps({"error_msg": traceback.format_exc()})
    return res


if __name__ == '__main__':
    a = 1
    # GetAssess()
    app.run(host="0.0.0.0", port="8082", threaded=True)