package com.example.services;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.controller.vo.DLModelVO;
import com.example.controller.vo.ModelVO;
import com.example.dao.*;
import com.example.data.FitTestData;
import com.example.data.LibsvmAdapter;
import com.example.entity.*;
import com.example.model.classification.*;
import com.example.model.cluster.KmeansCluster;
import com.example.model.helper.Utils;
import com.example.model.regression.RFRegression;
import com.example.services.pojo.DLGeneralModelPojo;
import com.example.services.pojo.DLModelPojo;
import com.example.services.pojo.ModelPojo;
import com.example.util.Json2Object;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.classification.LogisticRegressionModel;
import org.apache.spark.ml.clustering.KMeansModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Config-related service
 * Created by twenbo on 2016/11/21.
 */
@Slf4j
@Data
@Component
public class ModelService {

    @Autowired
    private ConfigDao configDao;
    @Autowired
    private FileInfoDao fileInfoDao;
    @Autowired
    private PictureFileDao pictureFileDao;
    @Autowired
    private ModelDao modelDao;
    @Autowired
    private ModelTypeDao modelTypeDao;
    @Autowired
    private DLModelDao dlModelDao;
    @Autowired
    private HeaderInfoDao headerInfoDao;
    @Autowired
    private LibsvmAdapter libsvmAdapter;
    @Autowired
    private DTClassification dtClassification;
    @Autowired
    private LRClassification lrClassification;
    @Autowired
    private RFClassification rfClassification;
    @Autowired
    private GDBTClassification gdbtClassification;
    @Autowired
    private KmeansCluster kmeansCluster;

    @Autowired
    private MulPerClassification mulPerClassification;
    @Autowired
    private NaiveBayesClassification naiveBayesClassification;

    @Autowired
    private RFRegression rfRegression;

    @Autowired
    private FitTestData fitTestData;
    @Autowired
    private FileService fileService;
    @Autowired
    private Utils utils;
    /**
     *
     * @param modelVO
     * @return
     */

    public int addModel(ModelVO modelVO){
        Model modelInfo = new Model();
        modelInfo.setFileInfoId(Integer.parseInt(modelVO.getFileInfoId()));
        modelInfo.setConfigId(Integer.parseInt(modelVO.getConfigId()));
        modelInfo.setModelName(modelVO.getModelName());
        modelInfo.setModelTypeId(Integer.parseInt(modelVO.getModelTypeId()));
        modelInfo.setTrained(false);
        modelInfo.setResOfModel(null);
        modelInfo.setModelPath(null);
        modelInfo.setArguments(JSON.toJSONString(modelVO.getArguments()));
        return modelDao.save(modelInfo).getId();
    }

    public int addDlModel(DLModelVO dlModelVO) {
        DLModel dlModel = new DLModel();
        dlModel.setModelTypeId(Integer.parseInt(dlModelVO.getModelTypeId()));
        dlModel.setPicFileId(Integer.parseInt(dlModelVO.getPicFileId()));
        dlModel.setModelName(dlModelVO.getModelName());
        dlModel.setTrained(false);
        dlModel.setResOfModel(null);
        dlModel.setModelPath(null);
        dlModel.setArguments(JSON.toJSONString(dlModelVO.getArguments()));
        return dlModelDao.save(dlModel).getId();
    }
    /**
     *
     * @param userId
     * @return
     */
    public List<ModelPojo> getAllModelByUserId(int userId){
        List<ModelPojo> rsl = configDao.findByUserId(userId).stream().map((config) -> {
            int configId = config.getId();
            String configName = config.getConfName();
            List<ModelPojo> modelPerConfig = modelDao.findByConfigId(configId).stream().map((model) -> {
                ModelType modelType =  modelTypeDao.findById(model.getModelTypeId());
                int flag = -1;
                if(model.isTrained() && model.getResOfModel()!=null){
                    flag = 1;
                }else if(model.isTrained() && model.getResOfModel()==null){
                    flag = 0;
                }
                return new ModelPojo(String.valueOf(configId),
                        configName,
                        String.valueOf(model.getModelTypeId()),
                        modelType != null ? modelType.getModelTypeName() : "",
                        modelType != null ? modelType.getModelDetailName() : "",
                        modelType != null ? modelType.getModelDetailDes() : "",
                        modelType != null ? modelType.getModelDes() : "",
                        String.valueOf(model.getId()),
                        model.getModelName(),
                        flag,
                        new Json2Object().Json2HashMap(model.getArguments()),
                        new Json2Object().Json2HashMap(model.getResOfModel()));
                }).collect(Collectors.toList());
            return modelPerConfig;
        }).reduce(new ArrayList<>(), (listA, listB) -> {
            listA.addAll(listB);
            return listA;
        });
        return rsl;
    }

    /**
     *
     * @param userId
     * @return
     */
    public List<DLGeneralModelPojo> getAllGeneralModelByUserId(int userId){
        List<PicFile> picFiles = pictureFileDao.findByUserId(userId);
        List<DLGeneralModelPojo> result = picFiles.stream().map(picFile -> {
            List<DLModel> dlModels = dlModelDao.findByPicFileId(picFile.getId());
            return dlModels.stream()
                    .map(dlModel -> new DLGeneralModelPojo(String.valueOf(dlModel.getId()), dlModel.getModelName(), String.valueOf(picFile.getId()), picFile.getFileName(),dlModel.isTrained()
                            ,dlModel.getResOfModel()!=null))
                    .collect(Collectors.toList());
        }).reduce(new ArrayList<>(), (listA, listB) -> {
            listA.addAll(listB);
            return listA;
        });
        return result;
    }

    public DLModelPojo getDLModelDetailById(int modelId) {
        DLModel dlModel = dlModelDao.findById(modelId);
        return new DLModelPojo(String.valueOf(dlModel.getModelTypeId()),String.valueOf(dlModel.getPicFileId()),
                dlModel.getModelName(),dlModel.isTrained(),dlModel.getModelPath(),new Json2Object().Json2HashMap(dlModel.getResOfModel())
        ,new Json2Object().Json2HashMap(dlModel.getArguments()));
    }

    /**
     *
     * @param modelTypeName
     * @return
     */
    public List<ModelType> getModelTypeByModelTypeName(String modelTypeName) {
        return modelTypeDao.findByModelTypeName(modelTypeName);
    }

    /**
     *
     * @param modelId
     * @param modelName
     * @param modelTypeId
     * @return
     */
    public boolean changeModel(int modelId, String modelName, int modelTypeId){
        Model model = modelDao.findById(modelId);
        model.setModelName(modelName);
        model.setModelTypeId(modelTypeId);
        modelDao.save(model);
        return true;
    }

    /**
     *change status of isTrained
     * @param modelId
     * @return
     */
    public boolean changeStatusOfTrained(int modelId, boolean trained){
        Model model =  modelDao.findById(modelId);
        model.setTrained(trained);
        modelDao.save(model);
        return true;
    }

    /**
     *
     * @param modelId
     * @param resOfModel
     * @param model
     * @return
     */
    public boolean updateResOfModel(int modelId, HashMap<String, String> resOfModel, Object model) throws Exception {

        // 将模型持久化（上传到hdfs）
        utils.storeModel(model,modelId);

        Model modelOfGet =  modelDao.findById(modelId);
        //log.info("\n\nmodel:"+JSON.toJSONString(model)+"\n\n");
        //T.B.D.
        //modelOfGet.setModel(model);
        modelOfGet.setResOfModel(JSON.toJSONString(resOfModel));
        modelOfGet.setTrained(true);
        log.info("\n\nstore model result: "+modelOfGet.getResOfModel());
        modelDao.save(modelOfGet);
        return true;
    }



    public boolean trainTheModel(int modelId) throws Exception{

        //change training flag to true
        changeStatusOfTrained(modelId, true);

        Model trainingModel = modelDao.findById(modelId);
        ModelType modelType = modelTypeDao.findById(trainingModel.getModelTypeId());
        Config configInfo = configDao.findById(trainingModel.getConfigId());
        List<String> features = JSONObject.parseArray(configInfo.getFieldIds(),String.class).stream().map(fieldId ->
                headerInfoDao.findById(Integer.parseInt(fieldId)).getFieldName()).collect(Collectors.toList());

        //1. transfer file to libsvm
        log.info("\n\nTransfer to LibSVM...\n\n");
        String keyFeature = features.get(features.size()-1);
        features.remove(features.size()-1);
        int featureNum = features.size();
        String libsvm = libsvmAdapter.parquet2Libsvm(configInfo.getFileInfoId(), keyFeature, features.toArray(new String[0]));

        String modelDetailName = modelType.getModelDetailName();
        HashMap<String,String> modelArguments = new Json2Object().Json2HashMap(trainingModel.getArguments());
        boolean rsl = false;
        try {
            switch (modelDetailName) {
                case "LogisticRegression":
            /*
            * maxIter: Int,
                 regParam: Float,
                 elasticNetParam: Float
            * */
                    LogisticRegressionModel lrModel = lrClassification.lrTraining(libsvm,
                            Integer.parseInt(modelArguments.get("MaxIter")),
                            Float.parseFloat(modelArguments.get("RegParam")),
                            Float.parseFloat(modelArguments.get("ElasticNetParam")));
                    log.info("\n\nstore Model result\n\n");
                    HashMap<String,String> lrRst = lrClassification.lrResult(lrModel,modelId);

                    log.info("\n\nupdate result of model\n\n");
                    rsl = updateResOfModel(modelId, lrRst, lrModel);
                    break;
                case "DecisionTree":
            /*
            * maxIter: Int,
                 regParam: Float,
                 elasticNetParam: Float
            * */
                    PipelineModel dtModel = dtClassification.dtTraining(libsvm,
                            Integer.parseInt(modelArguments.get("Category")),
                            Float.parseFloat(modelArguments.get("TrainingSetOccupy")));
                    HashMap<String,String> dtResult = dtClassification.dtResult(dtModel, modelId);

                    rsl = updateResOfModel(modelId, dtResult, dtModel);
                    break;
                case "RandomForest":
                    /*
                * maxIter: Int,
                 regParam: Float,
                 elasticNetParam: Float
            * */

                    PipelineModel rfModel = rfClassification.dtTraining(libsvm,
                            Integer.parseInt(modelArguments.get("Category")),
                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")));
                    HashMap<String,String> rfResult = rfClassification.dtResult(rfModel, modelId);

                    rsl = updateResOfModel(modelId,rfResult,rfModel);
                    break;
                case "GBDT":
                    /*
                * maxIter: Int,
                 regParam: Float,
                 elasticNetParam: Float
            * */

                    PipelineModel gbtModel = gdbtClassification.dtTraining(libsvm,
                            Integer.parseInt(modelArguments.get("Category")),
                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")));
                    HashMap<String,String> gbtResult = gdbtClassification.dtResult(gbtModel, modelId);

                    rsl = updateResOfModel(modelId,gbtResult,gbtModel);
                    break;

                case "K-Means":
                    KMeansModel kmeansModel = kmeansCluster.dtTraining(libsvm,
                            Integer.parseInt(modelArguments.get("NumOfCluster")),
                            Integer.parseInt(modelArguments.get("Seed")));
                    HashMap<String,String> kmeansResult = kmeansCluster.dtResult(kmeansModel,modelId);
                    rsl = updateResOfModel(modelId,kmeansResult,kmeansModel);
                    break;

                case "MultilayerPerceptronClassifier":
                    PipelineModel mlpc = mulPerClassification.dtTraining(libsvm,
                            Integer.parseInt(modelArguments.get("Category")),
                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")), featureNum);
                    HashMap<String,String> mlpcResult = mulPerClassification.dtResult(mlpc, modelId);
                    rsl = updateResOfModel(modelId,mlpcResult,mlpc);
                    break;

                case "NaiveBayes":
                    PipelineModel nby = naiveBayesClassification.dtTraining(libsvm,
                            Integer.parseInt(modelArguments.get("Category")),
                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy")));
                    HashMap<String,String> nbyResult = naiveBayesClassification.dtResult(nby, modelId);
                    rsl = updateResOfModel(modelId,nbyResult,nby);
                    break;

                case "RandomForestRegression":
                    PipelineModel rfReg = rfRegression.dtTraining(libsvm,
                            Integer.parseInt(modelArguments.get("Category")),
                            Float.parseFloat(modelArguments.get("TrainingDataSetOccupy"))
                            );
                    HashMap<String,String> rfRegResult = rfRegression.dtResult(rfReg,modelId);

                    rsl = updateResOfModel(modelId,rfRegResult,rfReg);
                    break;
                default:
                    rsl = false;
            }
        } catch (Exception e) {
            e.printStackTrace();
            changeStatusOfTrained(modelId, false);
        }
        if(!rsl){
            changeStatusOfTrained(modelId,false);
        }
        return rsl;
    }

    /**
     *
     * @param modelId
     */

    public boolean deleteModel(int modelId) {
        Model model =  modelDao.findById(modelId);
        if(model == null){
            return false;
        }
        modelDao.delete(modelDao.findById(modelId));
        return true;
    }

    public List<Map<String, String>> useModel(int fileId, int modelId) throws Exception{
        List<Map<String,String>> res = fitTestData.transformData(fileId,modelId);
//        fileService.deleteFile(fileId);
        return res;
    }


}