<filename>core/src/main/java/com/alibaba/alink/operator/batch/regression/GlmEvaluationBatchOp.java
package com.alibaba.alink.operator.batch.regression;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.ml.api.misc.param.Params;
import org.apache.flink.table.api.Table;
import org.apache.flink.types.Row;

import com.alibaba.alink.common.annotation.InputPorts;
import com.alibaba.alink.common.annotation.NameCn;
import com.alibaba.alink.common.annotation.OutputPorts;
import com.alibaba.alink.common.annotation.ParamSelectColumnSpec;
import com.alibaba.alink.common.annotation.PortSpec;
import com.alibaba.alink.common.annotation.PortType;
import com.alibaba.alink.common.annotation.TypeCollections;
import com.alibaba.alink.common.utils.DataSetConversionUtil;
import com.alibaba.alink.common.utils.JsonConverter;
import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.common.regression.glm.FamilyLink;
import com.alibaba.alink.operator.common.regression.glm.GlmUtil;
import com.alibaba.alink.operator.common.regression.glm.GlmUtil.GlmModelSummary;
import com.alibaba.alink.params.regression.GlmTrainParams;

/**
 * Generalized Linear Model. https://en.wikipedia.org/wiki/Generalized_linear_model.
 */
@InputPorts(values = {
	@PortSpec(PortType.MODEL),
	@PortSpec(PortType.DATA)
})
@OutputPorts(values = {
	@PortSpec(value = PortType.DATA)
})
@ParamSelectColumnSpec(name = "featureCols", allowedTypeCollections = TypeCollections.NUMERIC_TYPES)
@ParamSelectColumnSpec(name = "labelCol", allowedTypeCollections = TypeCollections.NUMERIC_TYPES)
@ParamSelectColumnSpec(name = "weightCol", allowedTypeCollections = TypeCollections.NUMERIC_TYPES)
@NameCn("广义线性回归评估")
public class GlmEvaluationBatchOp extends BatchOperator <GlmEvaluationBatchOp>
	implements GlmTrainParams <GlmEvaluationBatchOp> {

	private static final long serialVersionUID = 591674929382964676L;

	public GlmEvaluationBatchOp() {

	}

	public GlmEvaluationBatchOp(Params params) {
		super(params);
	}

	@Override
	public GlmEvaluationBatchOp linkFrom(BatchOperator <?>... inputs) {
		checkOpSize(2, inputs);

		BatchOperator <?> model = inputs[0];
		BatchOperator <?> in = inputs[1];

		String[] featureColNames = getFeatureCols();
		String labelColName = getLabelCol();
		String weightColName = getWeightCol();
		String offsetColName = getOffsetCol();

		Family familyName = getFamily();
		Link linkName = getLink();
		double variancePower = getVariancePower();
		double linkPower = getLinkPower();

		int numIter = getMaxIter();
		double epsilon = getEpsilon();

		boolean fitIntercept = getFitIntercept();
		double regParam = getRegParam();

		FamilyLink familyLink = new FamilyLink(familyName, variancePower, linkName, linkPower);

		int numFeature = featureColNames.length;

		DataSet <Row> data = GlmUtil.preProc(in, featureColNames, offsetColName, weightColName, labelColName);

		DataSet <GlmUtil.WeightedLeastSquaresModel> wlsModel = model.getDataSet().mapPartition(
			new GlmUtil.GlmModelToWlsModel());
		DataSet <Row> residual = GlmUtil.residual(wlsModel, data, numFeature, familyLink);
		DataSet <GlmModelSummary> aggSummay = GlmUtil.aggSummary(residual, wlsModel,
			numFeature, familyLink, regParam, numIter, epsilon, fitIntercept);

		//residual
		String[] residualColNames = new String[numFeature + 4 + 4];
		TypeInformation[] residualColTypes = new TypeInformation[numFeature + 4 + 4];
		for (int i = 0; i < numFeature; i++) {
			residualColNames[i] = featureColNames[i];
			residualColTypes[i] = Types.DOUBLE;
		}
		residualColNames[numFeature] = "label";
		residualColTypes[numFeature] = Types.DOUBLE;
		residualColNames[numFeature + 1] = "weight";
		residualColTypes[numFeature + 1] = Types.DOUBLE;
		residualColNames[numFeature + 2] = "offset";
		residualColTypes[numFeature + 2] = Types.DOUBLE;
		residualColNames[numFeature + 3] = "pred";
		residualColTypes[numFeature + 3] = Types.DOUBLE;
		residualColNames[numFeature + 4] = "residualdevianceResiduals";
		residualColTypes[numFeature + 4] = Types.DOUBLE;
		residualColNames[numFeature + 5] = "pearsonResiduals";
		residualColTypes[numFeature + 5] = Types.DOUBLE;
		residualColNames[numFeature + 6] = "workingResiduals";
		residualColTypes[numFeature + 6] = Types.DOUBLE;
		residualColNames[numFeature + 7] = "responseResiduals";
		residualColTypes[numFeature + 7] = Types.DOUBLE;

		this.setSideOutputTables(new Table[] {
			DataSetConversionUtil.toTable(getMLEnvironmentId(),
				residual, residualColNames, residualColTypes)
		});

		//summary
		String[] summaryColNames = new String[1];
		TypeInformation[] summaryColTypes = new TypeInformation[1];
		summaryColNames[0] = "summary";
		summaryColTypes[0] = Types.STRING;

		this.setOutput(aggSummay.map(
			new MapFunction <GlmModelSummary, Row>() {
				@Override
				public Row map(GlmModelSummary value) throws Exception {
					return Row.of(JsonConverter.toJson(value));
				}
			}), summaryColNames, summaryColTypes);

		return this;
	}
}
